mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-22 11:41:19 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

cmd/bootnode, eth, p2p, p2p/discover: clean up the seeder and mesh into eth.

Browse Source
This commit is contained in:
Péter Szilágyi 2015-04-24 11:19:33 +03:00
parent 971702e7a1
commit 6def110c37
9 changed files with 168 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/bootnode/main.go
View File

@ -71,7 +71,7 @@ func main() {
}
}
if _, err := discover.ListenUDP(nodeKey, *listenAddr, natm, ""); err != nil {
if _, err := discover.ListenUDP(nodeKey, *listenAddr, natm, nil); err != nil {
log.Fatal(err)
}
select {}

13
eth/backend.go
View File

@ -125,6 +125,8 @@ type Ethereum struct {
blockDb common.Database // Block chain database
stateDb common.Database // State changes database
extraDb common.Database // Extra database (txs, etc)
seedDb *discover.Cache // Peer database seeding the bootstrap
// Closed when databases are flushed and closed
databasesClosed chan bool
@ -179,7 +181,10 @@ func New(config *Config) (*Ethereum, error) {
if err != nil {
return nil, err
}
seedDbPath := path.Join(config.DataDir, "seeds")
seedDb, err := discover.NewPersistentCache(path.Join(config.DataDir, "seeds"))
if err != nil {
return nil, err
}
// Perform database sanity checks
d, _ := blockDb.Get([]byte("ProtocolVersion"))
@ -207,6 +212,7 @@ func New(config *Config) (*Ethereum, error) {
blockDb: blockDb,
stateDb: stateDb,
extraDb: extraDb,
seedDb: seedDb,
eventMux: &event.TypeMux{},
accountManager: config.AccountManager,
DataDir: config.DataDir,
@ -244,7 +250,7 @@ func New(config *Config) (*Ethereum, error) {
NAT: config.NAT,
NoDial: !config.Dial,
BootstrapNodes: config.parseBootNodes(),
SeedCache: seedDbPath,
SeedCache: seedDb,
}
if len(config.Port) > 0 {
eth.net.ListenAddr = ":" + config.Port
@ -423,6 +429,7 @@ done:
}
}
s.seedDb.Close()
s.blockDb.Close()
s.stateDb.Close()
s.extraDb.Close()
@ -450,7 +457,7 @@ func (self *Ethereum) SuggestPeer(nodeURL string) error {
}
func (s *Ethereum) Stop() {
s.txSub.Unsubscribe() // quits txBroadcastLoop
s.txSub.Unsubscribe() // quits txBroadcastLoop
s.protocolManager.Stop()
s.txPool.Stop()

134
p2p/discover/cache.go Normal file
View File

@ -0,0 +1,134 @@
// Contains the discovery cache, storing previously seen nodes to act as seed
// servers during bootstrapping the network.
package discover
import (
"bytes"
"encoding/binary"
"net"
"os"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/storage"
)
// Cache stores all nodes we know about.
type Cache struct {
db *leveldb.DB
}
// Cache version to allow dumping old data if it changes.
var cacheVersionKey = []byte("pv")
// NewMemoryCache creates a new in-memory peer cache without a persistent backend.
func NewMemoryCache() (*Cache, error) {
db, err := leveldb.Open(storage.NewMemStorage(), nil)
if err != nil {
return nil, err
}
return &Cache{db: db}, nil
}
// NewPersistentCache creates/opens a leveldb backed persistent peer cache, also
// flushing its contents in case of a version mismatch.
func NewPersistentCache(path string) (*Cache, error) {
// Try to open the cache, recovering any corruption
db, err := leveldb.OpenFile(path, nil)
if _, iscorrupted := err.(leveldb.ErrCorrupted); iscorrupted {
db, err = leveldb.RecoverFile(path, nil)
}
if err != nil {
return nil, err
}
// The nodes contained in the cache correspond to a certain protocol version.
// Flush all nodes if the version doesn't match.
currentVer := make([]byte, binary.MaxVarintLen64)
currentVer = currentVer[:binary.PutVarint(currentVer, Version)]
blob, err := db.Get(cacheVersionKey, nil)
switch err {
case leveldb.ErrNotFound:
// Version not found (i.e. empty cache), insert it
err = db.Put(cacheVersionKey, currentVer, nil)
case nil:
// Version present, flush if different
if !bytes.Equal(blob, currentVer) {
db.Close()
if err = os.RemoveAll(path); err != nil {
return nil, err
}
return NewPersistentCache(path)
}
}
// Clean up in case of an error
if err != nil {
db.Close()
return nil, err
}
return &Cache{db: db}, nil
}
// get retrieves a node with a given id from the seed da
func (c *Cache) get(id NodeID) *Node {
blob, err := c.db.Get(id[:], nil)
if err != nil {
return nil
}
node := new(Node)
if err := rlp.DecodeBytes(blob, node); err != nil {
return nil
}
return node
}
// list retrieves a batch of nodes from the database.
func (c *Cache) list(n int) []*Node {
it := c.db.NewIterator(nil, nil)
defer it.Release()
nodes := make([]*Node, 0, n)
for i := 0; i < n && it.Next(); i++ {
var id NodeID
copy(id[:], it.Key())
if node := c.get(id); node != nil {
nodes = append(nodes, node)
}
}
return nodes
}
// update inserts - potentially overwriting - a node in the seed database.
func (c *Cache) update(node *Node) error {
blob, err := rlp.EncodeToBytes(node)
if err != nil {
return err
}
return c.db.Put(node.ID[:], blob, nil)
}
// add inserts a new node into the seed database.
func (c *Cache) add(id NodeID, addr *net.UDPAddr, tcpPort uint16) *Node {
node := &Node{
ID: id,
IP: addr.IP,
DiscPort: addr.Port,
TCPPort: int(tcpPort),
}
c.update(node)
return node
}
// delete removes a node from the database.
func (c *Cache) delete(id NodeID) error {
return c.db.Delete(id[:], nil)
}
// Close flushes and closes the database files.
func (c *Cache) Close() {
c.db.Close()
}

114
p2p/discover/node.go
View File

@ -1,10 +1,8 @@
package discover
import (
"bytes"
"crypto/ecdsa"
"crypto/elliptic"
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
@ -13,16 +11,12 @@ import (
"math/rand"
"net"
"net/url"
"os"
"strconv"
"strings"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/secp256k1"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/opt"
"github.com/syndtr/goleveldb/leveldb/storage"
)
const nodeIDBits = 512
@ -310,111 +304,3 @@ func randomID(a NodeID, n int) (b NodeID) {
}
return b
}
// nodeDB stores all nodes we know about.
type nodeDB struct {
ldb *leveldb.DB
}
var dbVersionKey = []byte("pv")
// Opens the backing LevelDB. If path is "", we use an in-memory database.
func newNodeDB(path string, version int64) (db *nodeDB, err error) {
db = new(nodeDB)
opts := new(opt.Options)
if path == "" {
db.ldb, err = leveldb.Open(storage.NewMemStorage(), opts)
} else {
db.ldb, err = openNodeDB(path, opts, version)
}
return db, err
}
// openNodeDB opens a persistent seed cache, flushing old versions.
func openNodeDB(path string, opts *opt.Options, version int64) (*leveldb.DB, error) {
ldb, err := leveldb.OpenFile(path, opts)
if _, iscorrupted := err.(leveldb.ErrCorrupted); iscorrupted {
ldb, err = leveldb.RecoverFile(path, opts)
}
if err != nil {
return nil, err
}
// The nodes contained in the database correspond to a certain
// protocol version. Flush all nodes if the DB version doesn't match.
// There is no need to do this for memory databases because they
// won't ever be used with a different protocol version.
shouldVal := make([]byte, binary.MaxVarintLen64)
shouldVal = shouldVal[:binary.PutVarint(shouldVal, version)]
val, err := ldb.Get(dbVersionKey, nil)
if err == leveldb.ErrNotFound {
err = ldb.Put(dbVersionKey, shouldVal, nil)
} else if err == nil && !bytes.Equal(val, shouldVal) {
// Delete and start over.
ldb.Close()
if err = os.RemoveAll(path); err != nil {
return nil, err
}
return openNodeDB(path, opts, version)
}
if err != nil {
ldb.Close()
ldb = nil
}
return ldb, err
}
// get retrieves a node with a given id from the seed da
func (db *nodeDB) get(id NodeID) *Node {
v, err := db.ldb.Get(id[:], nil)
if err != nil {
return nil
}
n := new(Node)
if err := rlp.DecodeBytes(v, n); err != nil {
return nil
}
return n
}
// list retrieves a batch of nodes from the database.
func (db *nodeDB) list(n int) []*Node {
it := db.ldb.NewIterator(nil, nil)
defer it.Release()
nodes := make([]*Node, 0, n)
for i := 0; i < n && it.Next(); i++ {
var id NodeID
copy(id[:], it.Key())
if node := db.get(id); node != nil {
nodes = append(nodes, node)
}
}
return nodes
}
// update inserts - potentially overwriting - a node in the seed database.
func (db *nodeDB) update(n *Node) error {
v, err := rlp.EncodeToBytes(n)
if err != nil {
return err
}
return db.ldb.Put(n.ID[:], v, nil)
}
// add inserts a new node into the seed database.
func (db *nodeDB) add(id NodeID, addr *net.UDPAddr, tcpPort uint16) *Node {
n := &Node{ID: id, IP: addr.IP, DiscPort: addr.Port, TCPPort: int(tcpPort)}
db.update(n)
return n
}
// delete removes a node from the database.
func (db *nodeDB) delete(id NodeID) error {
return db.ldb.Delete(id[:], nil)
}
// close flushes and closes the database files.
func (db *nodeDB) close() {
db.ldb.Close()
}

23
p2p/discover/table.go
View File

@ -27,6 +27,7 @@ type Table struct {
mutex sync.Mutex // protects buckets, their content, and nursery
buckets [nBuckets]*bucket // index of known nodes by distance
nursery []*Node // bootstrap nodes
cache *Cache // cache of known nodes
bondmu sync.Mutex
bonding map[NodeID]*bondproc
@ -34,7 +35,6 @@ type Table struct {
net transport
self *Node // metadata of the local node
db *nodeDB
}
type bondproc struct {
@ -61,17 +61,15 @@ type bucket struct {
entries []*Node
}
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, seedCache string) *Table {
// Load the bootstrap seed cache (use in memory db upon failure)
db, err := newNodeDB(seedCache, Version)
if err != nil {
glog.V(logger.Warn).Infoln("Failed to open bootstrap seed cache:", err)
db, _ = newNodeDB("", Version)
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, seeder *Cache) *Table {
// If no seed cache was given, use an in-memory one
if seeder == nil {
seeder, _ = NewMemoryCache()
}
// Create the bootstrap table
tab := &Table{
net: t,
db: db,
cache: seeder,
self: newNode(ourID, ourAddr),
bonding: make(map[NodeID]*bondproc),
bondslots: make(chan struct{}, maxBondingPingPongs),
@ -93,7 +91,6 @@ func (tab *Table) Self() *Node {
// Close terminates the network listener and flushes the seed cache.
func (tab *Table) Close() {
tab.net.close()
tab.db.close()
}
// Bootstrap sets the bootstrap nodes. These nodes are used to connect
@ -178,10 +175,10 @@ func (tab *Table) refresh() {
result := tab.Lookup(randomID(tab.self.ID, ld))
if len(result) == 0 {
// Pick a batch of previously know seeds to lookup with and discard them (will come back if they are still live)
seeds := tab.db.list(10)
seeds := tab.cache.list(10)
for _, seed := range seeds {
glog.V(logger.Debug).Infoln("Seeding network with:", seed)
tab.db.delete(seed.ID)
tab.cache.delete(seed.ID)
}
// Bootstrap the table with a self lookup
all := tab.bondall(append(tab.nursery, seeds...))
@ -252,7 +249,7 @@ func (tab *Table) bondall(nodes []*Node) (result []*Node) {
// of the process can be skipped.
func (tab *Table) bond(pinged bool, id NodeID, addr *net.UDPAddr, tcpPort uint16) (*Node, error) {
var n *Node
if n = tab.db.get(id); n == nil {
if n = tab.cache.get(id); n == nil {
tab.bondmu.Lock()
w := tab.bonding[id]
if w != nil {
@ -297,7 +294,7 @@ func (tab *Table) pingpong(w *bondproc, pinged bool, id NodeID, addr *net.UDPAdd
// waitping will simply time out.
tab.net.waitping(id)
}
w.n = tab.db.add(id, addr, tcpPort)
w.n = tab.cache.add(id, addr, tcpPort)
close(w.done)
}

6
p2p/discover/table_test.go
View File

@ -15,7 +15,7 @@ import (
func TestTable_pingReplace(t *testing.T) {
doit := func(newNodeIsResponding, lastInBucketIsResponding bool) {
transport := newPingRecorder()
tab := newTable(transport, NodeID{}, &net.UDPAddr{}, "")
tab := newTable(transport, NodeID{}, &net.UDPAddr{}, nil)
last := fillBucket(tab, 200)
pingSender := randomID(tab.self.ID, 200)
@ -145,7 +145,7 @@ func TestTable_closest(t *testing.T) {
test := func(test *closeTest) bool {
// for any node table, Target and N
tab := newTable(nil, test.Self, &net.UDPAddr{}, "")
tab := newTable(nil, test.Self, &net.UDPAddr{}, nil)
tab.add(test.All)
// check that doClosest(Target, N) returns nodes
@ -217,7 +217,7 @@ func TestTable_Lookup(t *testing.T) {
self := gen(NodeID{}, quickrand).(NodeID)
target := randomID(self, 200)
transport := findnodeOracle{t, target}
tab := newTable(transport, self, &net.UDPAddr{}, "")
tab := newTable(transport, self, &net.UDPAddr{}, nil)
// lookup on empty table returns no nodes
if results := tab.Lookup(target); len(results) > 0 {

10
p2p/discover/udp.go
View File

@ -144,7 +144,7 @@ type reply struct {
}
// ListenUDP returns a new table that listens for UDP packets on laddr.
func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, seedCache string) (*Table, error) {
func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, seeder *Cache) (*Table, error) {
addr, err := net.ResolveUDPAddr("udp", laddr)
if err != nil {
return nil, err
@ -153,12 +153,12 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, seedCac
if err != nil {
return nil, err
}
tab, _ := newUDP(priv, conn, natm, seedCache)
tab, _ := newUDP(priv, conn, natm, seeder)
glog.V(logger.Info).Infoln("Listening,", tab.self)
return tab, nil
}
func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, seedCache string) (*Table, *udp) {
func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, seeder *Cache) (*Table, *udp) {
udp := &udp{
conn: c,
priv: priv,
@ -176,7 +176,7 @@ func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, seedCache string
realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port}
}
}
udp.Table = newTable(udp, PubkeyID(&priv.PublicKey), realaddr, seedCache)
udp.Table = newTable(udp, PubkeyID(&priv.PublicKey), realaddr, seeder)
go udp.loop()
go udp.readLoop()
return udp.Table, udp
@ -449,7 +449,7 @@ func (req *findnode) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte
if expired(req.Expiration) {
return errExpired
}
if t.db.get(fromID) == nil {
if t.cache.get(fromID) == nil {
// No bond exists, we don't process the packet. This prevents
// an attack vector where the discovery protocol could be used
// to amplify traffic in a DDOS attack. A malicious actor

4
p2p/discover/udp_test.go
View File

@ -41,7 +41,7 @@ func newUDPTest(t *testing.T) *udpTest {
remotekey: newkey(),
remoteaddr: &net.UDPAddr{IP: net.IP{1, 2, 3, 4}, Port: 30303},
}
test.table, test.udp = newUDP(test.localkey, test.pipe, nil, "")
test.table, test.udp = newUDP(test.localkey, test.pipe, nil, nil)
return test
}
@ -157,7 +157,7 @@ func TestUDP_findnode(t *testing.T) {
// ensure there's a bond with the test node,
// findnode won't be accepted otherwise.
test.table.db.add(PubkeyID(&test.remotekey.PublicKey), test.remoteaddr, 99)
test.table.cache.add(PubkeyID(&test.remotekey.PublicKey), test.remoteaddr, 99)
// check that closest neighbors are returned.
test.packetIn(nil, findnodePacket, &findnode{Target: testTarget, Expiration: futureExp})

6
p2p/server.go
View File

@ -59,9 +59,9 @@ type Server struct {
// with the rest of the network.
BootstrapNodes []*discover.Node
// SeedCache is the path to the database containing the previously seen live
// nodes in the network to use as potential bootstrap seeds.
SeedCache string
// SeedCache is the database containing the previously seen live nodes in
// the network to use as potential bootstrap seeds.
SeedCache *discover.Cache
// Protocols should contain the protocols supported
// by the server. Matching protocols are launched for