ความเข้าใจเชิงลึกเกี่ยวกับการออกแบบเครือข่าย Ethereum P2P จากชั้นโค้ด

创宇区块链安全实验室

特邀专栏作者

2021-09-24 19:20 (UTC+8)

บทความนี้มีประมาณ 35621 คำ การอ่านทั้งหมดใช้เวลาประมาณ 51 นาที

พื้นฐานของโครงสร้างเครือข่ายแบบกระจายบล็อกเชน เครือข่าย Ethereum P2P พัฒนาอย่างไร

สรุปโดย AI

คำนำ

บทความนี้จะให้ภาพรวมโดยย่อของการพัฒนาเครือข่าย P2P ในขณะเดียวกันก็จะแนะนำกลไกการค้นพบโหนด ตารางแฮชแบบกระจาย การค้นหาโหนด การเพิ่มโหนด และการลบโหนดใน Ethereum โดยสังเขปจากมุมมองของซอร์สโค้ด การวิเคราะห์โดยสังเขปของการออกแบบความปลอดภัยเครือข่าย P2P

ความรู้พื้นฐาน

ความรู้พื้นฐาน

เครือข่ายพีทูพี

เครือข่าย P2P แตกต่างจากโครงสร้าง CS แบบดั้งเดิม ในเครือข่าย P2P แต่ละโหนดสามารถเป็นไคลเอนต์หรือเซิร์ฟเวอร์ได้ โดยทั่วไป โปรโตคอลการสื่อสารระหว่างโหนดจะถูกนำไปใช้โดยตรงผ่านซ็อกเก็ต

การพัฒนาเทคโนโลยี P2P ได้ผ่านการพัฒนาสี่ขั้นตอนต่อไปนี้:

1. รวมศูนย์:

เป็นวิธีการกำหนดเส้นทางที่ง่ายที่สุดในโหมดเครือข่าย P2P นั่นคือมีโหนดกลางซึ่งเก็บข้อมูลดัชนีของโหนดอื่น ๆ ทั้งหมด ข้อมูลดัชนีโดยทั่วไปประกอบด้วยโหนด IP, พอร์ต, ทรัพยากรโหนด ฯลฯ ข้อดีของการรวมศูนย์ routing คือมีโครงสร้างที่เรียบง่ายและง่ายต่อการนำไปใช้ ง่าย แต่ข้อเสียก็ชัดเจนเช่นกัน เนื่องจากโหนดกลางจำเป็นต้องเก็บข้อมูลการเราต์ของทุกโหนด เมื่อขนาดของโหนดยังคงขยายตัว ปัญหาคอขวดของประสิทธิภาพมักจะเกิดขึ้น และยังมีปัญหาความล้มเหลวของโหนดเดียว

2. แจกจ่าย:

หมายความว่าโหนดกลางถูกลบออก, เครือข่ายแบบสุ่มถูกสร้างขึ้นระหว่างโหนด P2P และช่องทางการเชื่อมต่อถูกสร้างขึ้นแบบสุ่มระหว่างโหนดที่เพิ่มใหม่และโหนดในเครือข่าย P2P ดังนั้นการสร้างโครงสร้างโทโพโลยีแบบสุ่มเมื่อโหนดใหม่เข้าร่วม เครือข่ายจะสุ่มเลือกโหนดที่มีอยู่และสร้างความสัมพันธ์เพื่อนบ้าน หลังจากที่ node สร้างการเชื่อมต่อกับโหนดเพื่อนบ้านแล้วจำเป็นต้องกระจายสัญญาณเครือข่ายทั้งหมดเพื่อให้เครือข่ายทั้งหมดทราบการมีอยู่ของโหนด"ขั้นตอนการออกอากาศเฉพาะคือ: โหนดจะออกอากาศไปยังโหนดข้างเคียงก่อน และหลังจากที่โหนดข้างเคียงได้รับข้อความออกอากาศ โหนดจะออกอากาศต่อไปยังโหนดข้างเคียงของตนเอง และอื่น ๆ วิธีการออกอากาศนี้เรียกอีกอย่างว่า"กลไกน้ำท่วม

และปัญหาของกลไกการเกิดน้ำท่วมคือการควบคุมที่ไม่ดี ซึ่งส่วนใหญ่รวมถึงปัญหาใหญ่สองประการ: หนึ่งคือมันง่ายที่จะก่อตัวเป็นวงจรน้ำท่วม ตัวอย่างเช่น โหนด A ส่งข้อความไปยังโหนด B ไปยังโหนด C และโหนด C ออกอากาศ ไปยังโหนด C A เป็นวงจร ปัญหาอีกประการหนึ่งคือการรับช่วงของการตอบกลับข้อความ ตัวอย่างเช่น หากทรัพยากรที่โหนด A ต้องการขอนั้นมีหลายโหนดเป็นเจ้าของ ดังนั้น ในช่วงเวลาสั้น ๆ โหนดจำนวนมาก จะส่งข้อความตอบกลับไปยังโหนด A พร้อมกัน ซึ่งมีแนวโน้มที่จะทำให้โหนด A หยุดทำงานทันที

วิธีการกำจัดการวนรอบน้ำท่วมสามารถเรียนรู้ได้จากการควบคุมการแพร่ภาพน้ำท่วมในโปรโตคอลการกำหนดเส้นทางเครือข่าย IP วิธีหนึ่งคือตั้งค่า TTL สำหรับข้อความค้นหาแต่ละข้อความ ทุกครั้งที่ส่งต่อข้อความท่วมท้น ค่า TTL จะลดลง 1 เมื่อโหนดยอมรับ เมื่อ TTL เป็น 0 ข้อความจะไม่ถูกส่งต่อ ซึ่งสามารถหลีกเลี่ยงการวนซ้ำไม่สิ้นสุดของข้อความค้นหาในเครือข่าย และยังสามารถตั้งค่าสถานะเฉพาะสำหรับข้อความซ้ำ และจะไม่ส่งต่อข้อความที่ได้รับซ้ำ ข้อความเพื่อหลีกเลี่ยงการวนซ้ำไม่สิ้นสุดวิธีการตอบสนองต่อพายุสามารถทำการแบ่งส่วนเครือข่ายที่ชั้นดาต้าลิงค์เพื่อลดการกระจายข้อความข้ามเซกเมนต์

3. ไฮบริด:

ประเภทไฮบริดเป็นโครงสร้างแบบรวมศูนย์และแบบกระจาย มีเครือข่ายแบบกระจายจำนวนมากที่ประกอบด้วย super node ในเครือข่าย และแต่ละ super node มีโหนดธรรมดาหลายโหนดเพื่อสร้างเครือข่ายเฉพาะที่แบบรวมศูนย์ คุณสามารถเพิ่มโหนดธรรมดาใหม่ได้ก่อน เลือก super node เพื่อสื่อสาร และ super node จะพุชรายการ super node อื่น ๆ ไปยังโหนดที่เข้าร่วมใหม่ node ที่เข้าร่วมจะตัดสินใจเลือก super node เฉพาะเป็นโหนดหลักตามสถานะของ super node ในรายการ การออกอากาศแบบท่วมท้นของโครงสร้างนี้เกิดขึ้นระหว่างโหนดขั้นสูงสามารถหลีกเลี่ยงน้ำท่วมขนาดใหญ่ได้ ในการใช้งานจริง โครงสร้างไฮบริดเป็นสถาปัตยกรรมเครือข่ายที่ค่อนข้างยืดหยุ่นและมีประสิทธิภาพ และความยากในการนำไปใช้ค่อนข้างน้อย ดังนั้นระบบจำนวนมากในปัจจุบัน ตามโครงสร้างแบบผสมผสาน (Hybrid) พัฒนาและดำเนินการ

สี่ โครงสร้าง:

เครือข่าย P2P ที่มีโครงสร้างเป็นโครงสร้างเครือข่ายแบบกระจาย แตกต่างจากโครงสร้างแบบกระจายที่กล่าวถึงข้างต้น เครือข่ายแบบกระจายเป็นเครือข่ายแบบสุ่ม ในขณะที่เครือข่ายแบบมีโครงสร้างจะจัดระเบียบโหนดทั้งหมดในลักษณะที่เป็นระเบียบตามโครงสร้างบางอย่าง เช่น การสร้างเครือข่ายแบบวงแหวนหรือ เครือข่ายแบบต้นไม้ เครือข่ายที่มีโครงสร้างโดยทั่วไปจะขึ้นอยู่กับอัลกอริทึมตารางแฮชแบบกระจาย (Distributed Hash Table, DHI) ในการใช้งานเฉพาะ รูปแบบการใช้งานเฉพาะ ได้แก่ Chord, Pasty, CAN, Kademlia และอัลกอริทึมอื่นๆ

มีการเปรียบเทียบโครงสร้างเครือข่ายทั้งสี่ดังนี้:

การค้นพบโหนด

การค้นพบโหนดเป็นขั้นตอนแรกสำหรับโหนดใดๆ ในการเข้าถึงเครือข่าย P2P การค้นพบโหนดสามารถแบ่งออกเป็นสองประเภท:

- การค้นพบโหนดเริ่มต้น:

หมายถึงโหนดเป็นโหนดใหม่ที่ไม่เคยถูกเรียกใช้ โหนดไม่มี ข้อมูลใดๆ ของโหนดอื่นในเครือข่าย ขณะนี้ โหนดพบว่าพึ่งพาได้เฉพาะโหนดฮาร์ดโค้ดใน โหนดเพื่อรับข้อมูลของเครือข่าย P2P

- การค้นพบโหนดที่รู้จัก:

โหนดเคยทำงานมาก่อนและฐานข้อมูลโหนดจะบันทึกข้อมูลของโหนดอื่นในเครือข่าย ในขณะนี้ โหนดพบว่าสามารถใช้โหนดที่สรุปไว้ในฐานข้อมูลโหนดเพื่อรับข้อมูลของเครือข่าย P2P ดังนั้น ในการสร้างโทโพโลยีเครือข่ายของตัวเอง

โหนดเมล็ด"ในเครือข่าย P2P โหนดเริ่มต้นจะค้นพบโหนดอื่น ๆ ในเครือข่ายอย่างรวดเร็วผ่านโหนดที่เสถียรในระยะยาวบางโหนดเมื่อเริ่มต้น โหนดเหล่านี้เรียกว่า"โหนดเมล็ด

DNS-Seed：

(โดยทั่วไปแล้วข้อมูล seed node จะถูกฮาร์ดโค้ดไว้ในโค้ด) โดยทั่วไปแล้ว seed node สามารถแบ่งออกได้เป็นสองประเภทคือ"หรือที่เรียกว่า"โหนดเมล็ด DNS

IP-Seed：

, DNS เป็นบริการสืบค้นชื่อโดเมนที่ให้บริการโดยอินเทอร์เน็ต โดยจะแมปชื่อโดเมนและที่อยู่ IP และจัดเก็บไว้ในฐานข้อมูลแบบกระจาย เมื่อเราเข้าถึงเซิร์ฟเวอร์ DNS ให้ระบุชื่อโดเมน และเซิร์ฟเวอร์ DNS จะสอดคล้องกับ ส่งคืนที่อยู่ IP ของชื่อโดเมน

นั่นคือ ที่อยู่ IP ของโหนดเมล็ดจะถูกฮาร์ดโค้ดลงในโค้ด และที่อยู่ของโหนดที่ฮาร์ดโค้ดเหล่านี้เรียกว่า โหนดเมล็ด

อัลกอริทึม KDA

HashTable

Kademlia เป็นเทคโนโลยีตารางแฮชแบบกระจาย (DHT) เมื่อเปรียบเทียบกับเทคโนโลยี DHT อื่นๆ อัลกอริทึม KDA ใช้อัลกอริทึม XOR เพื่อคำนวณระยะห่างระหว่างโหนดและสร้างโทโพโลยี DHT ใหม่ อัลกอริทึมนี้สามารถปรับปรุงความเร็วในการสืบค้นของเส้นทางได้อย่างมาก .

ตารางแฮชเป็นคอนเทนเนอร์สำหรับจัดเก็บคู่คีย์-ค่า คู่คีย์-ค่า เรียกว่า คู่คีย์/ค่า โครงสร้างข้อมูลตารางแฮชมีที่เก็บข้อมูล N ชุด สำหรับตารางแฮชเฉพาะ N (จำนวนของที่เก็บข้อมูล) มักจะคงที่ คุณจึงสามารถกำหนดหมายเลขแต่ละบัคเก็ตได้ 0~N-1 และบัคเก็ตนี้ใช้เพื่อจัดเก็บคู่คีย์-ค่า เข้าใจง่ายๆ ว่าเป็นอาร์เรย์ไดนามิกซึ่งเก็บคู่คีย์-ค่าหลายคู่

HashCollision

รูปต่อไปนี้แสดงหลักการค้นหาของตารางแฮช เราสามารถรับค่าได้ง่ายและรวดเร็วผ่านคีย์ เมื่อใช้คีย์บางคีย์ในการค้นหา ขั้นแรกให้ใช้ฟังก์ชันแฮชเพื่อคำนวณค่าแฮชของคีย์ และค่าที่ได้ แฮช ค่ามักจะเป็นจำนวนเต็ม จากนั้นค่าแฮชจะถูกใช้เพื่อดำเนินการโมดูโลกับ N (จำนวนของบัคเก็ต) (ส่วนที่เหลือของการหาร) และสามารถคำนวณหมายเลขบัคเก็ตที่เกี่ยวข้องได้

เมื่อพูดถึงตารางแฮช เราต้องพูดถึงการชนกันของตารางแฮช เมื่อ 2 คีย์ที่ต่างกันทำการคำนวณแฮชเพื่อให้ได้ค่าแฮชที่เท่ากัน จะเรียกว่า การชนกันของฟังก์ชันแฮช เมื่อสิ่งนี้เกิดขึ้น คีย์ 2 คีย์ที่ตรงกัน 2 คีย์ คู่ค่าจะถูกจัดเก็บไว้ในบัคเก็ต (bucket) เดียวกัน การชนกันของแฮชอีกประการหนึ่งคือแม้ว่าค่าแฮชที่คำนวณได้จะแตกต่างกัน แต่จะได้หมายเลขบัคเก็ตเดียวกันหลังจากการดำเนินการโมดูโล บางครั้งคีย์-ค่าสองคู่จะถูกเก็บไว้ ในถัง หลักการชนกันของแฮชแสดงในรูปด้านล่าง:

ถ้าตารางแฮชไม่มีการชนกันเลยเมื่อจัดเก็บข้อมูล แสดงว่าในแต่ละบัคเก็ตจะมีคู่คีย์-ค่าเพียง 0 หรือ 1 คู่ ซึ่งทำให้ค้นหาได้อย่างรวดเร็ว ตรงกันข้าม หากตารางแฮชปรากฏขึ้นเมื่อจัดเก็บข้อมูลอย่างจริงจัง การชนกันจะทำให้คู่คีย์-ค่าหลายคู่ถูกจัดเก็บไว้ในบัคเก็ตหนึ่งๆ เมื่อค้นหาคีย์ จำเป็นต้องเปรียบเทียบว่าคีย์นั้นเหมือนกันในบัคเก็ตนี้ทีละคีย์หรือไม่ และประสิทธิภาพการค้นหาจะต่ำมาก~

ตารางแฮชแบบกระจาย

ตารางแฮชแบบกระจายมีแนวคิดคล้ายกับตารางแฮชแบบดั้งเดิม ความแตกต่างคือตารางแฮชแบบดั้งเดิมส่วนใหญ่จะใช้ในซอฟต์แวร์บางอย่างในเครื่องเดียว และตารางแฮชแบบกระจายส่วนใหญ่จะใช้ในระบบแบบกระจาย (ในขณะนี้ โหนดของระบบสามารถเข้าใจได้ทั่วไปว่าเป็นบัคเก็ตในตารางแฮช) ตารางแฮชแบบกระจายส่วนใหญ่จะใช้เพื่อเก็บข้อมูลจำนวนมาก (ถึงมหาศาล) หลักการของตารางแฮชแบบกระจายแสดงในรูปต่อไปนี้ :

การวิเคราะห์ซอร์สโค้ด

P2PServer พื้นฐานของ Ethereum สามารถแบ่งออกเป็นสามเลเยอร์ต่อไปนี้:

- ชั้นบนสุด:

การใช้งานเฉพาะของแต่ละโปรโตคอลใน Ethereum

- ชั้นกลาง:

ชั้นลิงค์การสื่อสาร p2p ใน Ethereum มีหน้าที่รับผิดชอบในการเริ่มตรวจสอบ ประมวลผลการเชื่อมต่อใหม่ หรือบำรุงรักษาการเชื่อมต่อ

- ชั้นล่าง:

เลเยอร์ IO ของเครือข่ายการสื่อสารข้อมูลใน Ethereum มีหน้าที่หลักในการจัดการตารางเส้นทางและการดำเนินการอ่านและเขียนของฐานข้อมูล

โครงสร้างตาราง

// 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 มีโหนด (เรียงลำดับจากมากไปน้อยตามกิจกรรมล่าสุด) ใช้เพื่อแสดงรายการดัชนีโหนดที่รู้จักตามระยะทาง
เนอสเซอรี่: โหนดเมล็ด เมื่อโหนดเริ่มทำงาน จะสามารถเชื่อมโยงโหนดเมล็ดได้สูงสุด 35 โหนด โดย 5 โหนดถูกกำหนดอย่างเป็นทางการโดย Ethereum และอีก 30 โหนดถูกแยกออกจากฐานข้อมูล
db: ฐานข้อมูลที่ใช้ในการจัดเก็บโหนด P2P (มีสองโหนดใน Ethereum และอีกอันหนึ่งใช้เพื่อเก็บข้อมูลในห่วงโซ่)
refreshReq: ไปป์ไลน์เพื่อรีเฟรชเหตุการณ์ K-bucket

การสร้างตาราง

// 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 โหนด ซึ่งจะใช้ในภายหลังหากตารางว่างเปล่าหรือ ไม่มีข้อมูลโหนดในฐานข้อมูลเพื่อเชื่อมโยงเครือข่าย) จากนั้นใช้ for loop รวมกับฟังก์ชัน ValidateComplete เพื่อตรวจสอบว่าโหนดนั้นถูกต้องหรือไม่

for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}

จากนั้นเริ่มต้นถัง K:

tab.seedRand()
tab.loadSeedNodes()
return tab, nil

จากนั้นสุ่มเลือก 30 โหนดจาก table.buckets และโหลดโหนดเมล็ดไปยังบัคเก็ตที่เกี่ยวข้อง:

การใช้งานเฉพาะของฟังก์ชัน 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 ลูปที่นี่จะรีเฟรชบัคเก็ต K ทุก 30 นาทีโดยอัตโนมัติ ตรวจสอบว่าโหนด K-bucket สามารถ ping ได้ทุก 10 วินาทีหรือไม่ และจัดเก็บโหนดใน K-bucket นานกว่า 5 นาทีในฐานข้อมูลในเครื่องทุกๆ 30 วินาทีเป็นโหนดที่เสถียร:
tab.seedRand()
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
กรณี <-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 bucket
for _, ch := range waiting {
close(ch)
}
waiting, refreshDone = nil, nil
กรณี <-refreshDone: // doRefresh เสร็จสมบูรณ์
revalidateDone = make(chan struct{})
go tab.doRevalidate(revalidateDone)
กรณี <-revalidate.C: // ตรวจสอบความถูกต้องของโหนด k-bucket 10 วินาที
revalidate.Reset(tab.nextRevalidateTime())
revalidateDone = nil
กรณี <-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()
}
}

เครือข่ายแบบกระจาย Ethereum ใช้โมเดลเครือข่ายที่มีโครงสร้างและโครงร่างการใช้งานใช้โปรโตคอล Kademlia ด้านล่างเราจะแนะนำการค้นพบโหนดโดยย่อ ใน Ethereum ค่าของ k คือ 16 ซึ่งหมายความว่าแต่ละ k bucket มี 16 โหนด 256 k-buckets , K-buckets จะบันทึก 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{})

ขั้นแรกให้ตรวจสอบว่าโหนดปัจจุบันอยู่ในสถานะกำลังทำงานหรือไม่ หากใช่ ให้ส่งคืนโดยตรงและแสดงข้อความแสดงข้อผิดพลาด หากไม่ใช่ ให้ตั้งค่า srv.running เป็น true จากนั้นเข้าสู่กระบวนการเริ่มต้นบริการ จากนั้นตรวจสอบว่าเปิดใช้งานบันทึกหรือไม่ ฯลฯ จากนั้นเริ่มต้นการกำหนดค่าข้อมูลบริการ P2P:

จากนั้นเรียก setupLocalNode เพื่อเริ่มโหนดภายในเครื่อง และสร้างจอภาพภายในเครื่อง จากนั้นกำหนดค่าโปรโตคอลเครือข่าย DiscoveryV5 เพื่อสร้างตารางเส้นทางโหนด

srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}

จากนั้นเรียก setupDialScheduler เพื่อเริ่มกระบวนการเชื่อมต่อผ่านสายโทรศัพท์ที่ใช้งานอยู่ จากนั้นเปิด coroutine เพื่อรักษาเพียร์:

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() ซึ่งเป็น goroutine แบบวนซ้ำไม่สิ้นสุด ซึ่งจะฟังพอร์ตและรับคำขอจากภายนอก:

// 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) ถูกเรียกใช้อีกครั้ง ซึ่งใช้เพื่อดำเนินการโปรโตคอล handshake:
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,ถ้า dialDest != ไม่มี { // 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
)
รหัส Handshake มีดังต่อไปนี้ โดยที่ตรรกะของ Handshake ที่สอดคล้องกันจะถูกดำเนินการโดยพิจารณาว่าเป็นการจับมือแบบแอ็คทีฟหรือการจับมือแบบพาสซีฟ:
sec, err = initiatorEncHandshake(c.conn, prv,ถ้า c.dialDest != ไม่มี { // จับมือกัน
c.dialDest) // เริ่มต้นการยืนยันคีย์แบบเชิงรุก สิ้นสุดการจับมือกัน ยืนยันคีย์ที่ใช้ร่วมกัน
sec, err = receiverEncHandshake(c.conn, prv)
}
if err != nil {
return nil, err
}
c.InitWithSecrets(sec)
return sec.remote, err
}

} อื่น { // จับมือเรื่อย ๆ

// 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 รวมถึงลายเซ็น รหัสสาธารณะ และหมายเลขเวอร์ชันที่สอดคล้องกับลายเซ็นที่สร้างขึ้นโดย nonce แบบสุ่ม จากนั้นเมธอด 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)
}

ตัวรับ EncHandshake มีดังต่อไปนี้ ซึ่งเกือบจะเหมือนกับตัว initialorEncHandshake:

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 โทรส่งที่นี่เพื่อส่งการดำเนินการจับมือ จากนั้นอ่านข้อมูลที่ส่งคืนผ่าน 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

ETH

ยินดีต้อนรับเข้าร่วมชุมชนทางการของ Odaily