코드 레이어에서 Ethereum P2P 네트워크 설계에 대한 심층적 이해
머리말
머리말
이 글은 P2P 네트워크의 발전 과정을 간략하게 소개함과 동시에 이더리움의 노드 탐색 메커니즘, 분산 해시 테이블, 노드 조회, 노드 추가 및 노드 제거를 소스 코드의 관점에서 간략하게 소개합니다. P2P 네트워크 보안 설계에 대한 간략한 분석.
기본 지식
기본 지식
P2P 네트워크
P2P 네트워크는 전통적인 CS 구조와 달리 P2P 네트워크에서 각 노드는 클라이언트 또는 서버가 될 수 있으며 노드 간의 통신 프로토콜은 일반적으로 소켓을 통해 직접 구현됩니다.
P2P 기술의 발전은 다음과 같은 4단계의 발전을 거쳤습니다.
1. 중앙화:
P2P 네트워크 모드에서 가장 간단한 라우팅 방식 즉, 중앙 노드가 있고 다른 모든 노드의 인덱스 정보를 저장하고 있으며 인덱스 정보에는 일반적으로 노드 IP, 포트, 노드 리소스 등이 포함됩니다. 라우팅은 구조가 간단하고 구현하기 쉽다 간편하지만 단점도 분명하다 중앙 노드가 모든 노드의 라우팅 정보를 저장해야 하므로 노드 규모가 계속 확장되면 성능 병목 현상이 발생하기 쉽다 , 단일 노드 장애 문제도 있습니다.
2. 분산:
중앙 노드가 제거되고 P2P 노드 간에 임의의 네트워크가 형성되며, 새로 추가된 노드와 P2P 네트워크의 노드 간에 연결 채널이 임의로 설정되어 임의의 토폴로지 구조를 형성하는 것을 의미합니다. 네트워크에서 임의로 기존 노드를 선택하고 이웃 관계를 설정합니다.노드가 이웃 노드와 연결을 설정한 후 전체 네트워크에 노드의 존재를 알리기 위해 전체 네트워크를 브로드캐스트해야 합니다."특정 브로드캐스트 단계는 다음과 같습니다. 노드가 먼저 이웃 노드에 브로드캐스트하고 이웃 노드가 브로드캐스트 메시지를 수신한 후 자신의 이웃 노드에 계속 브로드캐스트하는 등의 브로드캐스트 방법이라고도 합니다."범람 메커니즘
플러딩 메커니즘의 문제는 제어 가능성이 낮다는 것입니다. 여기에는 주로 두 가지 주요 문제가 포함됩니다. 하나는 플러딩 사이클을 형성하기 쉽다는 것입니다. 예를 들어 노드 A는 노드 B에서 노드 C로 메시지를 보내고 노드 C는 브로드 캐스트합니다. 또 다른 문제는 메시지 응답의 하도급 문제인데, 예를 들어 A노드가 요청하고자 하는 자원을 여러 노드가 소유하고 있다면 단기간에 많은 수의 노드가 노드 A에 동시에 응답 메시지를 보내므로 노드 A가 즉시 충돌할 수 있습니다.
플러딩 루프를 제거하는 방법은 IP 네트워크 라우팅 프로토콜의 플러딩 브로드캐스트 제어에서 배울 수 있습니다.한 가지 방법은 각 쿼리 메시지에 대해 TTL 값을 설정하는 것입니다.플러딩 메시지가 전달될 때마다 TTL 값이 1씩 감소합니다. 노드가 수락할 때 TTL이 0일 때 메시지가 전달되지 않으므로 네트워크에서 쿼리 메시지의 무한 루프를 피할 수 있으며 플러드 메시지에 대한 고유 플래그를 설정할 수 있으며 수신된 중복을 전달하지 않습니다. 폭풍우에 대응하는 방법은 데이터 링크 계층에서 네트워크 분할을 수행하여 메시지의 교차 세그먼트 방송을 줄일 수 있습니다.
3. 하이브리드:
하이브리드형은 사실 집중형과 분산형이 혼합된 구조로, 네트워크에는 슈퍼노드로 구성된 분산 네트워크가 많고 각 슈퍼노드는 여러 개의 일반 노드를 가지고 있어 로컬 집중형 네트워크를 형성한다.새로운 일반 노드를 먼저 추가할 수 있다. .통신할 슈퍼노드를 선택하면 슈퍼노드는 다른 슈퍼노드 목록을 새로운 가입 노드로 푸시합니다.가입 노드는 목록의 슈퍼노드 상태에 따라 특정 슈퍼노드를 부모 노드로 선택하기로 결정합니다. 이 구조의 플러드 브로드캐스트는 슈퍼 노드 사이에서 대규모 플러딩을 피할 수 있습니다.실제 응용에서 하이브리드 구조는 비교적 유연하고 효과적인 네트워킹 아키텍처이며 구현 난이도가 비교적 작습니다.따라서 현재 많은 시스템이 하이브리드 구조를 기반으로 개발 및 구현
4. 구조화:
구조화된 P2P 네트워크는 분산형 네트워크 구조로서 위에서 언급한 분산형 구조와 달리 임의의 네트워크인 반면, 구조형 네트워크는 모든 노드를 일정한 구조에 따라 규칙적으로 구성하여 Ring 네트워크를 형성하거나 트리 네트워크, 구조화된 네트워크는 일반적으로 특정 구현에서 분산 해시 테이블(Distributed Hash Table, DHI) 알고리즘을 기반으로 하며 특정 구현 체계에는 Chord, Pasty, CAN, Kademlia 및 기타 알고리즘이 포함됩니다.
네 가지 네트워크 구조는 다음과 같이 비교됩니다.
노드 검색
노드 탐색은 모든 노드가 P2P 네트워크에 액세스하기 위한 첫 번째 단계로 노드 탐색은 두 가지 유형으로 나눌 수 있습니다.
- 초기 노드 검색:
노드는 한 번도 실행된 적이 없는 새로운 노드를 의미합니다. 노드는 네트워크의 다른 노드에 대한 데이터가 없습니다. 이 때 노드는 하드 코딩된 시드 노드에만 의존할 수 있음을 발견합니다. P2P 네트워크의 정보를 얻기 위한 노드.
- 알려진 노드 검색:
노드는 이전에 실행되었으며 노드 데이터베이스는 네트워크의 다른 노드의 정보를 저장합니다.이 때 노드는 P2P 네트워크의 정보를 얻기 위해 노드 데이터베이스에 요약된 노드에 의존할 수 있음을 발견합니다. 자체 네트워크 토폴로지를 구축하기 위해
시드 노드"P2P 네트워크에서 초기 노드는 시작할 때 장기적으로 안정적인 일부 노드를 통해 네트워크의 다른 노드를 빠르게 발견합니다."시드 노드
DNS-Seed：
(일반적으로 시드 노드 정보는 코드에 하드 코딩됩니다.) 일반적으로 시드 노드는 두 가지 유형으로 나눌 수 있습니다."또한 ~으로 알려진"DNS 시드 노드
IP-Seed：
, DNS는 인터넷에서 제공하는 도메인 이름 조회 서비스입니다. 도메인 이름과 IP 주소를 매핑하여 분산 데이터베이스에 저장합니다. DNS 서버에 액세스할 때 도메인 이름을 제공하면 DNS 서버가 해당됩니다. 반환된 도메인 이름 IP 주소
즉, 시드노드의 IP 주소를 코드에 하드코딩하고, 이렇게 하드코딩된 노드의 주소를 시드노드라고 합니다.
KDA 알고리즘
HashTable
Kademlia는 분산 해시 테이블(DHT) 기술입니다.KDA 알고리즘은 다른 DHT 기술과 비교하여 XOR 알고리즘을 사용하여 노드 간의 거리를 계산한 다음 새로운 DHT 토폴로지를 설정합니다.이 알고리즘은 경로의 쿼리 속도를 크게 향상시킬 수 있습니다. .
해시 테이블은 키-값 쌍을 저장하는 컨테이너입니다. 키-값 쌍을 키/값 쌍이라고 합니다. 해시 테이블 데이터 구조는 N개의 버킷을 포함합니다. 특정 해시 테이블의 경우 일반적으로 N(버킷 수)이 고정되어 있습니다. 각 버킷에 0~N-1까지 번호를 매길 수 있으며, 버킷은 키-값 쌍을 저장하는 데 사용되며 간단히 여러 개의 키-값 쌍을 저장하는 동적 배열로 이해할 수 있습니다.
HashCollision
다음 그림은 해시 테이블의 검색 원리를 보여줍니다.키를 통해 쉽고 빠르게 값을 얻을 수 있습니다.특정 키를 사용하여 검색할 때 먼저 특정 해시 함수를 사용하여 키의 해시 값을 계산하고 얻은 해시 값은 일반적으로 정수이며 해시 값은 N(버킷 수)(나누기의 나머지)에 대해 모듈로 연산을 수행하는 데 사용되며 해당 버킷 수를 계산할 수 있습니다.
해시 테이블에 관해서는 해시 테이블 충돌을 언급해야 합니다.두 개의 서로 다른 키가 동일한 해시 값을 얻기 위해 해시 계산을 수행하는 것을 소위 해시 함수 충돌이라고 합니다.이렇게 되면 두 개의 키가 해당 두 개의 키 -value pair가 같은 버킷(bucket)에 저장됨 또 다른 해시 충돌은 계산된 해시 값이 다르지만 modulo 연산 후 동일한 버킷 번호를 획득하는 경우.때때로 두 개의 키-값 쌍이 저장됨 해시 충돌 원리는 아래 그림과 같습니다.
해시 테이블이 데이터를 저장할 때 충돌이 전혀 없다면 각 버킷에는 0 또는 1개의 키-값 쌍만 있으므로 조회가 매우 빠릅니다.반대로 데이터를 저장할 때 해시 테이블이 나타나면 심각함 충돌로 인해 많은 키-값 쌍이 특정 버킷에 저장됩니다.키를 검색할 때 이 버킷에서 키가 동일한지 하나씩 비교해야 하며 검색 효율이 매우 낮아집니다~
분산 해시 테이블
분산 해시 테이블은 개념적으로 전통적인 해시 테이블과 유사하지만, 차이점은 기존 해시 테이블은 단일 시스템의 특정 소프트웨어에서 주로 사용되며 분산 해시 테이블은 분산 시스템(이때, 분산 시스템의 노드는 일반적으로 해시 테이블의 버킷으로 이해할 수 있습니다.) 분산 해시 테이블은 주로 많은 양의 데이터를 저장하는 데 사용됩니다. 분산 해시 테이블의 원리는 다음 그림과 같습니다. :
소스 코드 분석
Ethereum의 기본 P2PServer는 대략 다음 세 계층으로 나눌 수 있습니다.
- 최상층:
이더리움에서 각 프로토콜의 특정 구현
- 중간층:
Ethereum의 p2p 통신 링크 계층은 모니터링 시작, 새 연결 처리 또는 연결 유지를 담당합니다.
- 최하층:
Ethereum의 데이터 통신 네트워크의 IO 계층은 주로 라우팅 테이블 관리와 데이터베이스의 읽기 및 쓰기 작업을 담당합니다.
테이블 구조
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L40
const (
alpha = 3 // Kademlia concurrency factor
bucketSize = 16 // Kademlia bucket size
maxReplacements = 10 // Size of per-bucket replacement list
// We keep buckets for the upper 1/15 of distances because
// it's very unlikely we'll ever encounter a node that's closer.
hashBits = len(common.Hash{}) * 8
nBuckets = hashBits / 15 // Number of buckets
bucketMinDistance = hashBits - nBuckets // Log distance of closest bucket
// IP address limits.
bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24
tableIPLimit, tableSubnet = 10, 24
refreshInterval = 30 * time.Minute
revalidateInterval = 10 * time.Second
copyNodesInterval = 30 * time.Second
seedMinTableTime = 5 * time.Minute
seedCount = 30
seedMaxAge = 5 * 24 * time.Hour
)
type Table struct {
mutex sync.Mutex // protects buckets, bucket content, nursery, rand
buckets [nBuckets]*bucket // index of known nodes by distance
nursery []*node // bootstrap nodes
rand *mrand.Rand // source of randomness, periodically reseeded
ips netutil.DistinctNetSet
log log.Logger
db *enode.DB // database of known nodes
net transport
refreshReq chan chan struct{}
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
nodeAddedHook func(*node) // for testing
}
type bucket struct {
entries []*node // live entries, sorted by time of last contact
replacements []*node // recently seen nodes to be used if revalidation fails
ips netutil.DistinctNetSet
}
테이블 데이터 구조는 다음과 같습니다.
몇 가지 주요 변수:
버킷: K 버킷, 각 K 버킷에는 거리별로 알려진 노드 인덱스를 나열하는 데 사용되는 노드(최근 활동에 따라 내림차순으로 정렬됨)가 포함됩니다.
Nursery: 시드 노드, 노드가 시작되면 최대 35개의 시드 노드를 연결할 수 있으며 그 중 5개는 Ethereum에서 공식적으로 지정하고 나머지 30개는 데이터베이스에서 추출됩니다.
db: P2P 노드를 저장하는 데 사용되는 데이터베이스(이더리움에 두 개가 있고 다른 하나는 체인에 데이터를 저장하는 데 사용됨)
- refreshReq: K-버킷 이벤트를 새로 고치는 파이프라인
테이블 생성
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L102
func newTable(t transport, db *enode.DB, bootnodes []*enode.Node, log log.Logger) (*Table, error) {
tab := &Table{
net: t,
db: db,
refreshReq: make(chan chan struct{}),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
closed: make(chan struct{}),
rand: mrand.New(mrand.NewSource(0)),
ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit},
log: log,
}
if err := tab.setFallbackNodes(bootnodes); err != nil {
return nil, err
}
for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
tab.seedRand()
tab.loadSeedNodes()
return tab, nil
}
newTable 함수는 새 테이블을 만드는 데 사용됩니다.
위의 코드에서 테이블 개체 탭은 먼저 들어오는 매개 변수로 초기화되고 초기 링크 노드를 설정하기 위해 setFallbackNodes 함수가 호출됩니다(즉, 테이블이 비어 있거나 나중에 사용되는 5개의 간호사 노드를 얻기 위해). 데이터베이스에 노드 정보가 없습니다. 네트워크 연결), ValidateComplete 함수와 결합된 for 루프를 사용하여 노드가 유효한지 확인합니다.
for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
그런 다음 K 버킷을 초기화합니다.
tab.seedRand()
tab.loadSeedNodes()
return tab, nil
그런 다음 table.buckets에서 임의로 30개의 노드를 선택하고 시드 노드를 해당 버킷에 로드합니다.
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L302
func (tab *Table) loadSeedNodes() {
seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
seeds = append(seeds, tab.nursery...)
for i := range seeds {
seed := seeds[i]
age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
tab.addSeenNode(seed)
}
}
loadSeedNodes 함수의 구체적인 구현은 다음과 같습니다(여기서 seedCount는 table.go의 맨 위에 정의된 전역 변수이며 값은 30입니다).
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L458
func (tab *Table) addSeenNode(n *node) {
if n.ID() == tab.self().ID() {
return
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
if contains(b.entries, n.ID()) {
// Already in bucket, don't add.
return
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, n)
return
}
if !tab.addIP(b, n.IP()) {
// Can't add: IP limit reached.
return
}
// Add to end of bucket:
b.entries = append(b.entries, n)
b.replacements = deleteNode(b.replacements, n)
n.addedAt = time.Now()
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(n)
}
}
여기서 addSeenNode는 버킷에 노드를 추가하는 데 사용됩니다. 여기에서 추가할 노드가 이미 존재하는지, 버킷이 가득 찼는지 여부를 확인합니다. 가득 찼다면 tab.addReplacement(b, n)를 호출하여 노드를 추가합니다. 교체 목록에 그런 다음 IP를 추가하고 버킷을 업데이트합니다.
이벤트 리스너
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L55
refreshInterval = 30 * time.Minute
revalidateInterval = 10 * time.Second
copyNodesInterval = 30 * time.Second
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L218
// loop schedules runs of doRefresh, doRevalidate and copyLiveNodes.
func (tab *Table) loop() {
var (
revalidate = time.NewTimer(tab.nextRevalidateTime())
refresh = time.NewTicker(refreshInterval)
copyNodes = time.NewTicker(copyNodesInterval)
refreshDone = make(chan struct{}) // where doRefresh reports completion
revalidateDone chan struct{} // where doRevalidate reports completion
waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
)
defer refresh.Stop()
defer revalidate.Stop()
defer copyNodes.Stop()
// Start initial refresh.
go tab.doRefresh(refreshDone)
loop:
for {
select {
루프 함수는 table.go의 메인 루프입니다. 나중에 사용할 로컬 변수를 함수 시작 부분에 정의한 다음 deRefresh를 통해 버킷을 새로 고칩니다. 여기서 루프는 30분마다 K 버킷을 자동으로 새로 고칩니다. K-bucket 노드가 10초마다 ping할 수 있는지 확인하고 30초마다 안정적인 노드로 로컬 데이터베이스에 5분 이상 K-bucket의 노드를 저장합니다.
tab.seedRand()
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case <-refresh.C: //k 버킷 이벤트를 정기적으로 새로고침, refreshInterval=30분
waiting = append(waiting, req)
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case req := <-tab.refreshReq: //k 버킷의 요청 이벤트 새로 고침
for _, ch := range waiting {
close(ch)
}
waiting, refreshDone = nil, nil
case <-refreshDone: // doRefresh 완료
revalidateDone = make(chan struct{})
go tab.doRevalidate(revalidateDone)
case <-revalidate.C: // k-bucket 노드의 유효성 검증, 10초
revalidate.Reset(tab.nextRevalidateTime())
revalidateDone = nil
case <-revalidateDone: // K 버킷 노드의 유효성 검증 완료
go tab.copyLiveNodes()
case <-tab.closeReq:
break loop
}
}
if refreshDone != nil {
<-refreshDone
}
for _, ch := range waiting {
close(ch)
}
if revalidateDone != nil {
<-revalidateDone
}
close(tab.closed)
case <-copyNodes.C: // 주기적으로 노드를 데이터베이스에 저장(30초) k-bucket에 노드가 5분 이상 존재하면 안정적인 노드로 간주
} 노드 조회
// getNode returns the node with the given ID or nil if it isn't in the table.
func (tab *Table) getNode(id enode.ID) *enode.Node {
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(id)
for _, e := range b.entries {
if e.ID() == id {
return unwrapNode(e)
}
}
return nil
}
getNode 함수는 ID로 노드를 찾는 데 사용되며 존재하지 않는 경우 nil을 반환합니다.
노드 검색
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L278
func (tab *Table) doRefresh(done chan struct{}) {
defer close(done)
tab.loadSeedNodes()
// Run self lookup to discover new neighbor nodes.
tab.net.lookupSelf()
for i := 0; i < 3; i++ {
tab.net.lookupRandom()
}
}
이더리움 분산 네트워크는 구조화된 네트워크 모델을 채택하고 그 구현 방식은 Kademlia 프로토콜을 사용합니다.아래에서 노드 탐색을 간략하게 소개합니다.이더리움에서 k 값은 16이며, 이는 각 k 버킷에 16개의 노드가 포함됨을 의미합니다.256k-버킷 , K-bucket은 노드의 NodeId, Distance, Endpoint, IP 및 기타 정보를 기록하고 Target 노드로부터의 거리에 따라 정렬합니다.노드 검색은 doRefresh()에 의해 구현됩니다.
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L302
func (tab *Table) loadSeedNodes() {
seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
seeds = append(seeds, tab.nursery...)
for i := range seeds {
seed := seeds[i]
age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
tab.addSeenNode(seed)
}
}
// filedir:go-ethereum-1.10.2\p2p\enode\nodedb.go L440
// QuerySeeds retrieves random nodes to be used as potential seed nodes
// for bootstrapping.
func (db *DB) QuerySeeds(n int, maxAge time.Duration) []*Node {
var (
now = time.Now()
nodes = make([]*Node, 0, n)
it = db.lvl.NewIterator(nil, nil)
id ID
)
defer it.Release()
seek:
for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ {
ctr := id[0]
rand.Read(id[:])
id[0] = ctr + id[0]%16
it.Seek(nodeKey(id))
n := nextNode(it)
if n == nil {
id[0] = 0
continue seek // iterator exhausted
}
if now.Sub(db.LastPongReceived(n.ID(), n.IP())) > maxAge {
continue seek
}
for i := range nodes {
if nodes[i].ID() == n.ID() {
continue seek // duplicate
}
}
nodes = append(nodes, n)
}
return nodes
}
위의 코드에서 볼 수 있듯이 먼저 tab.loadSeedNodes()를 호출하여 데이터베이스에서 노드를 로드하고 테이블에 삽입합니다.
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L281
// lookupSelf looks up our own node ID.
// This is needed to satisfy the transport interface.
func (t *UDPv5) lookupSelf() []*enode.Node {
return t.newLookup(t.closeCtx, t.Self().ID()).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L292
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
return t.lookupWorker(n, target)
})
}
func newLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *lookup {
it := &lookup{
tab: tab,
queryfunc: q,
asked: make(map[enode.ID]bool),
seen: make(map[enode.ID]bool),
result: nodesByDistance{target: target},
replyCh: make(chan []*node, alpha),
cancelCh: ctx.Done(),
queries: -1,
}
// Don't query further if we hit ourself.
// Unlikely to happen often in practice.
it.asked[tab.self().ID()] = true
return it
}
그런 다음 lookupSelf를 사용하여 새 노드를 검색합니다. 여기에서 현재 노드의 ID는 먼저 newLookup을 실행하여 이웃 노드를 검색하는 데 사용됩니다.
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L275
func (t *UDPv5) lookupRandom() []*enode.Node {
return t.newRandomLookup(t.closeCtx).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L287
func (t *UDPv5) newRandomLookup(ctx context.Context) *lookup {
var target enode.ID
crand.Read(target[:])
return t.newLookup(ctx, target)
}
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
return t.lookupWorker(n, target)
})
}
// lookupWorker performs FINDNODE calls against a single node during lookup.
func (t *UDPv5) lookupWorker(destNode *node, target enode.ID) ([]*node, error) {
var (
dists = lookupDistances(target, destNode.ID())
nodes = nodesByDistance{target: target}
err error
)
var r []*enode.Node
r, err = t.findnode(unwrapNode(destNode), dists)
if err == errClosed {
return nil, err
}
for _, n := range r {
if n.ID() != t.Self().ID() {
nodes.push(wrapNode(n), findnodeResultLimit)
}
}
return nodes.entries, err
}
// filedir:go-ethereum-1.10.2\p2p\discover\v5_udp.go L362
// findnode calls FINDNODE on a node and waits for responses.
func (t *UDPv5) findnode(n *enode.Node, distances []uint) ([]*enode.Node, error) {
resp := t.call(n, v5wire.NodesMsg, &v5wire.Findnode{Distances: distances})
return t.waitForNodes(resp, distances)
}
func (t *UDPv5) call(node *enode.Node, responseType byte, packet v5wire.Packet) *callV5 {
c := &callV5{
node: node,
packet: packet,
responseType: responseType,
reqid: make([]byte, 8),
ch: make(chan v5wire.Packet, 1),
err: make(chan error, 1),
}
// Assign request ID.
crand.Read(c.reqid)
packet.SetRequestID(c.reqid)
// Send call to dispatch.
select {
case t.callCh <- c:
case <-t.closeCtx.Done():
c.err <- errClosed
}
return c
}
마지막으로 조회를 위해 임의의 대상이 수행됩니다.
서비스 구조
// filedir:go-ethereum-1.10.2\p2p\server.go L160
// Server manages all peer connections.
type Server struct {
// Config fields may not be modified while the server is running.
Config
// Hooks for testing. These are useful because we can inhibit
// the whole protocol stack.
newTransport func(net.Conn, *ecdsa.PublicKey) transport
newPeerHook func(*Peer)
listenFunc func(network, addr string) (net.Listener, error)
lock sync.Mutex // protects running
running bool
listener net.Listener
ourHandshake *protoHandshake
loopWG sync.WaitGroup // loop, listenLoop
peerFeed event.Feed
log log.Logger
nodedb *enode.DB
localnode *enode.LocalNode
ntab *discover.UDPv4
DiscV5 *discover.UDPv5
discmix *enode.FairMix
dialsched *dialScheduler
// Channels into the run loop.
quit chan struct{}
addtrusted chan *enode.Node
removetrusted chan *enode.Node
peerOp chan peerOpFunc
peerOpDone chan struct{}
delpeer chan peerDrop
checkpointPostHandshake chan *conn
checkpointAddPeer chan *conn
// State of run loop and listenLoop.
inboundHistory expHeap
}
서버 측의 데이터 구조는 다음과 같습니다.
// Config holds Server options.
type Config struct {
PrivateKey *ecdsa.PrivateKey `toml:"-"`
MaxPeers int
MaxPendingPeers int `toml:",omitempty"`
DialRatio int `toml:",omitempty"`
NoDiscovery bool
DiscoveryV5 bool `toml:",omitempty"`
Name string `toml:"-"`
BootstrapNodes []*enode.Node
BootstrapNodesV5 []*enode.Node `toml:",omitempty"`
StaticNodes []*enode.Node
TrustedNodes []*enode.Node
NetRestrict *netutil.Netlist `toml:",omitempty"`
NodeDatabase string `toml:",omitempty"`
Protocols []Protocol `toml:"-"`
ListenAddr string
NAT nat.Interface `toml:",omitempty"`
Dialer NodeDialer `toml:"-"`
NoDial bool `toml:",omitempty"`
EnableMsgEvents bool
Logger log.Logger `toml:",omitempty"`
clock mclock.Clock
}
#####
서버 구성(로컬 노드 비밀 키, 전화 접속 비율, 최대 노드 연결 수, 전화 접속 비율, 이벤트 레코드 등):
새 노드
// filedir:go-ethereum-1.10.2\p2p\server.go L318
func (srv *Server) AddPeer(node *enode.Node) {
srv.dialsched.addStatic(node)
}
// filedir:go-ethereum-1.10.2\p2p\dial.go L190
// addStatic adds a static dial candidate.
func (d *dialScheduler) addStatic(n *enode.Node) {
select {
case d.addStaticCh <- n:
case <-d.ctx.Done():
}
}
AddPeer 함수는 주어진 노드를 추가하는 데 사용되며 구현 코드는 다음과 같습니다.
// AddTrustedPeer adds the given node to a reserved whitelist which allows the
// node to always connect, even if the slot are full.
func (srv *Server) AddTrustedPeer(node *enode.Node) {
select {
case srv.addtrusted <- node:
case <-srv.quit:
}
}
AddTrustedPeer 함수는 신뢰할 수 있는 노드를 추가하는 데 사용됩니다.
노드 제거
// filedir:go-ethereum-1.10.2\p2p\server.go L325
func (srv *Server) RemovePeer(node *enode.Node) {
var (
ch chan *PeerEvent
sub event.Subscription
)
// Disconnect the peer on the main loop.
srv.doPeerOp(func(peers map[enode.ID]*Peer) {
srv.dialsched.removeStatic(node)
if peer := peers[node.ID()]; peer != nil {
ch = make(chan *PeerEvent, 1)
sub = srv.peerFeed.Subscribe(ch)
peer.Disconnect(DiscRequested)
}
})
// Wait for the peer connection to end.
if ch != nil {
defer sub.Unsubscribe()
for ev := range ch {
if ev.Peer == node.ID() && ev.Type == PeerEventTypeDrop {
return
}
}
}
}
// filedir:go-ethereum-1.10.2\p2p\dial.go L198
// removeStatic removes a static dial candidate.
func (d *dialScheduler) removeStatic(n *enode.Node) {
select {
case d.remStaticCh <- n:
case <-d.ctx.Done():
}
}
RemovePeer 함수는 노드를 제거하고 노드에서 연결을 끊는 데 사용됩니다.
// RemoveTrustedPeer removes the given node from the trusted peer set.
func (srv *Server) RemoveTrustedPeer(node *enode.Node) {
select {
case srv.removetrusted <- node:
case <-srv.quit:
}
}
RemoveTrustedPeer 함수는 신뢰할 수 있는 피어를 제거하는 데 사용됩니다.
서비스 종료
func (srv *Server) Stop() {
srv.lock.Lock()
if !srv.running {
srv.lock.Unlock()
return
}
srv.running = false
if srv.listener != nil {
// this unblocks listener Accept
srv.listener.Close()
}
close(srv.quit)
srv.lock.Unlock()
srv.loopWG.Wait()
}
Stop 기능은 노드 동작을 종료하는 데 사용되며 구체적인 코드는 다음과 같습니다.
서비스 시작
// filedir:go-ethereum-1.10.2\p2p\server.go L433
func (srv *Server) Start() (err error) {
srv.lock.Lock()
defer srv.lock.Unlock()
if srv.running {
return errors.New("server already running")
}
srv.running = true
srv.log = srv.Config.Logger
if srv.log == nil {
srv.log = log.Root()
}
if srv.clock == nil {
srv.clock = mclock.System{}
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
// static fields
if srv.PrivateKey == nil {
return errors.New("Server.PrivateKey must be set to a non-nil key")
}
if srv.newTransport == nil {
srv.newTransport = newRLPX
}
if srv.listenFunc == nil {
srv.listenFunc = net.Listen
}
srv.quit = make(chan struct{})
srv.delpeer = make(chan peerDrop)
srv.checkpointPostHandshake = make(chan *conn)
srv.checkpointAddPeer = make(chan *conn)
srv.addtrusted = make(chan *enode.Node)
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
if err := srv.setupLocalNode(); err != nil {
return err
}
if srv.ListenAddr != "" {
if err := srv.setupListening(); err != nil {
return err
}
}
if err := srv.setupDiscovery(); err != nil {
return err
}
srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}
go-ethereum-1.10.2\p2p\server.go에 있는 시작 기능은 P2P 노드를 시작하는 데 사용됩니다.
// Start starts running the server.
// Servers can not be re-used after stopping.
func (srv *Server) Start() (err error) {
srv.lock.Lock()
defer srv.lock.Unlock()
if srv.running {
return errors.New("server already running")
}
srv.running = true
srv.log = srv.Config.Logger
if srv.log == nil {
srv.log = log.Root()
}
if srv.clock == nil {
srv.clock = mclock.System{}
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
// static fields
if srv.PrivateKey == nil {
return errors.New("Server.PrivateKey must be set to a non-nil key")
}
if srv.newTransport == nil {
srv.newTransport = newRLPX
}
if srv.listenFunc == nil {
srv.listenFunc = net.Listen
}
srv.quit = make(chan struct{})
srv.delpeer = make(chan peerDrop)
srv.checkpointPostHandshake = make(chan *conn)
srv.checkpointAddPeer = make(chan *conn)
srv.addtrusted = make(chan *enode.Node)
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
여기서 현재 노드가 running 상태인지 먼저 확인하고, 있다면 바로 리턴하여 에러 메시지를 주고, 그렇지 않다면 srv.running을 true로 설정한 후 서비스 시작 프로세스에 진입한 후 로그 활성화 여부를 확인한다. 등을 입력한 다음 구성 P2P 서비스 정보를 초기화합니다.
그런 다음 setupLocalNode를 호출하여 로컬 노드를 시작하고 로컬 모니터를 설정한 다음 DiscoveryV5 네트워크 프로토콜을 구성하여 노드 라우팅 테이블을 생성합니다.
srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}
그런 다음 setupDialScheduler를 호출하여 활성 전화 접속 연결 프로세스를 시작한 다음 피어를 유지 관리할 코루틴을 엽니다.
func (srv *Server) setupDialScheduler() {
config := dialConfig{
self: srv.localnode.ID(),
maxDialPeers: srv.maxDialedConns(),
maxActiveDials: srv.MaxPendingPeers,
log: srv.Logger,
netRestrict: srv.NetRestrict,
dialer: srv.Dialer,
clock: srv.clock,
}
if srv.ntab != nil {
config.resolver = srv.ntab
}
if config.dialer == nil {
config.dialer = tcpDialer{&net.Dialer{Timeout: defaultDialTimeout}}
}
srv.dialsched = newDialScheduler(config, srv.discmix, srv.SetupConn)
for _, n := range srv.StaticNodes {
srv.dialsched.addStatic(n)
}
}
setupDialScheduler 코드는 다음과 같습니다. 여기에서 연결은 newDialScheduler를 통해 설정되고, 매개 변수 discmix는 능동적으로 연결을 설정할 때 노드 세트를 결정하며 반복자이며 동시에 setupConn 연결 설정 기능이 전달됩니다.
// filedir:go-ethereum-1.10.2\p2p\dial.go L162
func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupFunc) *dialScheduler {
d := &dialScheduler{
dialConfig: config.withDefaults(),
setupFunc: setupFunc,
dialing: make(map[enode.ID]*dialTask),
static: make(map[enode.ID]*dialTask),
peers: make(map[enode.ID]connFlag),
doneCh: make(chan *dialTask),
nodesIn: make(chan *enode.Node),
addStaticCh: make(chan *enode.Node),
remStaticCh: make(chan *enode.Node),
addPeerCh: make(chan *conn),
remPeerCh: make(chan *conn),
}
d.lastStatsLog = d.clock.Now()
d.ctx, d.cancel = context.WithCancel(context.Background())
d.wg.Add(2)
go d.readNodes(it)
go d.loop(it)
return d
}
newDialScheduler 함수는 다음과 같습니다. 여기서 노드는 d.readNodes(it)를 통해 반복자에서 가져온 다음 연결을 위해 채널을 통해 d.loop(it)로 전달됩니다.
서비스 모니터링
func (srv *Server) setupListening() error {
// Launch the listener.
listener, err := srv.listenFunc("tcp", srv.ListenAddr)
if err != nil {
return err
}
srv.listener = listener
srv.ListenAddr = listener.Addr().String()
// Update the local node record and map the TCP listening port if NAT is configured.
if tcp, ok := listener.Addr().(*net.TCPAddr); ok {
srv.localnode.Set(enr.TCP(tcp.Port))
if !tcp.IP.IsLoopback() && srv.NAT != nil {
srv.loopWG.Add(1)
go func() {
nat.Map(srv.NAT, srv.quit, "tcp", tcp.Port, tcp.Port, "ethereum p2p")
srv.loopWG.Done()
}()
}
}
srv.loopWG.Add(1)
go srv.listenLoop()
return nil
}
위의 서비스 시작 과정에는 이벤트를 수신하는 setupListening 함수가 있으며 구체적인 코드는 다음과 같습니다.
// listenLoop runs in its own goroutine and accepts
// inbound connections.
func (srv *Server) listenLoop() {
srv.log.Debug("TCP listener up", "addr", srv.listener.Addr())
// The slots channel limits accepts of new connections.
tokens := defaultMaxPendingPeers
if srv.MaxPendingPeers > 0 {
tokens = srv.MaxPendingPeers
}
slots := make(chan struct{}, tokens)
for i := 0; i < tokens; i++ {
slots <- struct{}{}
}
// Wait for slots to be returned on exit. This ensures all connection goroutines
// are down before listenLoop returns.
defer srv.loopWG.Done()
defer func() {
for i := 0; i < cap(slots); i++ {
<-slots
}
}()
for {
// Wait for a free slot before accepting.
<-slots
var (
fd net.Conn
err error
lastLog time.Time
)
for {
fd, err = srv.listener.Accept()
if netutil.IsTemporaryError(err) {
if time.Since(lastLog) > 1*time.Second {
srv.log.Debug("Temporary read error", "err", err)
lastLog = time.Now()
}
time.Sleep(time.Millisecond * 200)
continue
} else if err != nil {
srv.log.Debug("Read error", "err", err)
slots <- struct{}{}
return
}
break
}
remoteIP := netutil.AddrIP(fd.RemoteAddr())
if err := srv.checkInboundConn(remoteIP); err != nil {
srv.log.Debug("Rejected inbound connection", "addr", fd.RemoteAddr(), "err", err)
fd.Close()
slots <- struct{}{}
continue
}
if remoteIP != nil {
var addr *net.TCPAddr
if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok {
addr = tcp
}
fd = newMeteredConn(fd, true, addr)
srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr())
}
go func() {
srv.SetupConn(fd, inboundConn, nil)
slots <- struct{}{}
}()
}
}
위의 코드에서 또 다른 srv.listenLoop()가 호출되는데, 이는 포트를 수신하고 외부 요청을 수신하는 무한 루프 고루틴입니다.
// SetupConn runs the handshakes and attempts to add the connection
// as a peer. It returns when the connection has been added as a peer
// or the handshakes have failed.
func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) error {
c := &conn{fd: fd, flags: flags, cont: make(chan error)}
if dialDest == nil {
c.transport = srv.newTransport(fd, nil)
} else {
c.transport = srv.newTransport(fd, dialDest.Pubkey())
}
err := srv.setupConn(c, flags, dialDest)
if err != nil {
c.close(err)
}
return err
}
여기서 SetupConn은 주로 핸드셰이크 프로토콜을 실행하고 링크를 피어 개체로 생성하려고 시도합니다.
func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) error {
// Prevent leftover pending conns from entering the handshake.
srv.lock.Lock()
running := srv.running
srv.lock.Unlock()
if !running {
return errServerStopped
}
// If dialing, figure out the remote public key.
var dialPubkey *ecdsa.PublicKey
위의 코드에서 핸드셰이크 프로토콜을 실행하는 데 사용되는 srv.setupConn(c, flags, dialDest) 함수가 다시 호출됩니다.
dialPubkey = new(ecdsa.PublicKey)
if err := dialDest.Load((*enode.Secp256k1)(dialPubkey)); err != nil {
err = errors.New("dial destination doesn't have a secp256k1 public key")
srv.log.Trace("Setting up connection failed", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
return err
}
}
// Run the RLPx handshake.
remotePubkey,if dialDest != nil { // dest=nil 수동 연결, dest != nil 활성 연결
if err != nil {
srv.log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
return err
}
if dialDest != nil {
c.node = dialDest
} else {
c.node = nodeFromConn(remotePubkey, c.fd)
}
clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags)
err = srv.checkpoint(c, srv.checkpointPostHandshake)
if err != nil {
clog.Trace("Rejected peer", "err", err)
return err
}
// Run the capability negotiation handshake.
phs,err := c.doEncHandshake(srv.PrivateKey) // 공개 키 교환, 공유 비밀 키의 RLPx 수준에서 핸드셰이크 결정
if err != nil {
clog.Trace("Failed p2p handshake", "err", err)
return err
}
if id := c.node.ID(); !bytes.Equal(crypto.Keccak256(phs.ID), id[:]) {
clog.Trace("Wrong devp2p handshake identity", "phsid", hex.EncodeToString(phs.ID))
return DiscUnexpectedIdentity
}
c.caps, c.name = phs.Caps, phs.Name
err = srv.checkpoint(c,err := c.doProtoHandshake(srv.ourHandshake) // 프로토콜 수준에서의 핸드셰이크, 즉 p2p 핸드셰이크, 왔다 갔다
if err != nil {
clog.Trace("Rejected peer", "err", err)
return err
}
return nil
}
srv.checkpointAddPeer) // 상태 확인
// filedir:go-ethereum-1.10.2\p2p\transport.go L123
func (t *rlpxTransport) doEncHandshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
t.conn.SetDeadline(time.Now().Add(handshakeTimeout))
return t.conn.Handshake(prv)
}
키 핸드셰이크는 deEncHandshake 함수를 통해 구현되며 Handshake() 함수는 함수에서 호출됩니다.
// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go L253
func (c *Conn) Handshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
var (
sec Secrets
err error
)
핸드셰이크 코드는 다음과 같습니다. 여기서 해당 핸드셰이크 논리는 능동 핸드셰이크인지 수동 핸드셰이크인지에 따라 실행됩니다.
sec, err = initiatorEncHandshake(c.conn, prv,if c.dialDest != nil { //액티브 핸드셰이크
c.dialDest) // 키 확인 핸드셰이크 종료를 사전에 시작하고 공유 키를 확인합니다.
sec, err = receiverEncHandshake(c.conn, prv)
}
if err != nil {
return nil, err
}
c.InitWithSecrets(sec)
return sec.remote, err
}
} else { // 수동 핸드셰이크
// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go L477
func initiatorEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey, remote *ecdsa.PublicKey) (s Secrets, err error) {
h := &encHandshake{initiator: true, remote: ecies.ImportECDSAPublic(remote)}
authMsg, err := h.makeAuthMsg(prv)
if err != nil {
return s, err
}
authPacket, err := sealEIP8(authMsg, h)
if err != nil {
return s, err
}
if _, err = conn.Write(authPacket); err != nil {
return s, err
}
authRespMsg := new(authRespV4)
authRespPacket, err := readHandshakeMsg(authRespMsg, encAuthRespLen, prv, conn)
if err != nil {
return s, err
}
if err := h.handleAuthResp(authRespMsg); err != nil {
return s, err
}
return h.secrets(authPacket, authRespPacket)
}
핸드셰이크를 능동적으로 시작하는 과정은 다음과 같다 여기서 makeAuthMsg를 호출하여 서명, 랜덤 난스에 의해 생성된 서명에 해당하는 공개 키 및 버전 번호를 포함하는 Auth 식별 정보를 생성한 후 sealEIP8 메서드는 다음과 같다. rlpx 인코딩을 수행하기 위해 호출된 다음 암호화된 핸드셰이크가 시작된 다음 authResp 메시지를 반환하고 확인 및 해독하고 상대방의 공개 키를 얻은 다음 AES, MAC을 생성합니다.
func receiverEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey) (s Secrets, err error) {
authMsg := new(authMsgV4)
authPacket, err := readHandshakeMsg(authMsg, encAuthMsgLen, prv, conn)
if err != nil {
return s, err
}
h := new(encHandshake)
if err := h.handleAuthMsg(authMsg, prv); err != nil {
return s, err
}
authRespMsg, err := h.makeAuthResp()
if err != nil {
return s, err
}
var authRespPacket []byte
if authMsg.gotPlain {
authRespPacket, err = authRespMsg.sealPlain(h)
} else {
authRespPacket, err = sealEIP8(authRespMsg, h)
}
if err != nil {
return s, err
}
if _, err = conn.Write(authRespPacket); err != nil {
return s, err
}
return h.secrets(authPacket, authRespPacket)
}
receiverEncHandshake는 다음과 같으며 initiatorEncHandshake와 거의 동일합니다.
func (t *rlpxTransport) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err error) {
werr := make(chan error, 1)
go func() { werr <- Send(t, handshakeMsg, our) }()
if their, err = readProtocolHandshake(t); err != nil {
<-werr // make sure the write terminates too
return nil, err
}
if err := <-werr; err != nil {
return nil, fmt.Errorf("write error: %v", err)
}
// If the protocol version supports Snappy encoding, upgrade immediately
t.conn.SetSnappy(their.Version >= snappyProtocolVersion)
return their, nil
}
그런 다음 doProtoHandshake를 통해 프로토콜 핸드셰이크 작업을 완료하고 여기에서 send를 호출하여 핸드셰이크 작업을 보낸 다음 readProtocolHandshake를 통해 반환된 정보를 읽고 다음을 확인합니다.
서비스 주기
func (srv *Server) run() {
srv.log.Info("Started P2P networking", "self", srv.localnode.Node().URLv4())
defer srv.loopWG.Done()
defer srv.nodedb.Close()
defer srv.discmix.Close()
defer srv.dialsched.stop()
var (
peers = make(map[enode.ID]*Peer)
inboundCount = 0
trusted = make(map[enode.ID]bool, len(srv.TrustedNodes))
)
for _, n := range srv.TrustedNodes {
trusted[n.ID()] = true
}
running:
for {
select {
case <-srv.quit:
break running
case n := <-srv.addtrusted:
srv.log.Trace("Adding trusted node", "node", n)
trusted[n.ID()] = true
if p, ok := peers[n.ID()]; ok {
p.rw.set(trustedConn, true)
}
case n := <-srv.removetrusted:
srv.log.Trace("Removing trusted node", "node", n)
delete(trusted, n.ID())
if p, ok := peers[n.ID()]; ok {
p.rw.set(trustedConn, false)
}
case op := <-srv.peerOp:
op(peers)
srv.peerOpDone <- struct{}{}
case c := <-srv.checkpointPostHandshake:
if trusted[c.node.ID()] {
c.flags |= trustedConn
}
c.cont <- srv.postHandshakeChecks(peers, inboundCount, c)
case c := <-srv.checkpointAddPeer:
err := srv.addPeerChecks(peers, inboundCount, c)
if err == nil {
p := srv.launchPeer(c)
peers[c.node.ID()] = p
srv.log.Debug("Adding p2p peer", "peercount", len(peers), "id", p.ID(), "conn", c.flags, "addr", p.RemoteAddr(), "name", p.Name())
srv.dialsched.peerAdded(c)
if p.Inbound() {
inboundCount++
}
}
c.cont <- err
case pd := <-srv.delpeer:
// A peer disconnected.
d := common.PrettyDuration(mclock.Now() - pd.created)
delete(peers, pd.ID())
srv.log.Debug("Removing p2p peer", "peercount", len(peers), "id", pd.ID(), "duration", d, "req", pd.requested, "err", pd.err)
srv.dialsched.peerRemoved(pd.rw)
if pd.Inbound() {
inboundCount--
}
}
}
srv.log.Trace("P2P networking is spinning down")
if srv.ntab != nil {
srv.ntab.Close()
}
if srv.DiscV5 != nil {
srv.DiscV5.Close()
}
for _, p := range peers {
p.Disconnect(DiscQuitting)
}
for len(peers) > 0 {
p := <-srv.delpeer
p.log.Trace("<-delpeer (spindown)")
delete(peers, p.ID())
}
}
실행 기능은 서버 종료 모니터링, 트러스트 노드 추가, 트러스트 노드 제거, 체크 노드 추가 등 서비스의 메인 루프입니다.
노드 정보
// NodeInfo gathers and returns a collection of metadata known about the host.
func (srv *Server) NodeInfo() *NodeInfo {
node := srv.Self()
info := &NodeInfo{
Name: srv.Name,
Enode: node.URLv4(),
ID: node.ID().String(),
IP: &nb
