From b05f0240d48ba0c0c3f9cb1fec948b05733f2ba3 Mon Sep 17 00:00:00 2001 From: Sonic Date: Tue, 25 Sep 2018 16:53:56 +0800 Subject: dex: implement notary node info propagation and management mechanism --- dex/handler.go | 164 ++++++++++++++++++++++++++++++++++++++++++++++++----- dex/helper_test.go | 24 +++++++- dex/protocol.go | 11 +++- 3 files changed, 183 insertions(+), 16 deletions(-) diff --git a/dex/handler.go b/dex/handler.go index 21609a561..96d20b02b 100644 --- a/dex/handler.go +++ b/dex/handler.go @@ -59,6 +59,11 @@ var ( // not compatible (low protocol version restrictions and high requirements). var errIncompatibleConfig = errors.New("incompatible configuration") +type newNotarySetEvent struct { + Round uint64 + Set map[string]struct{} +} + func errResp(code errCode, format string, v ...interface{}) error { return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...)) } @@ -74,9 +79,10 @@ type ProtocolManager struct { chainconfig *params.ChainConfig maxPeers int - downloader *downloader.Downloader - fetcher *fetcher.Fetcher - peers *peerSet + downloader *downloader.Downloader + fetcher *fetcher.Fetcher + peers *peerSet + notarySetManager *notarySetManager SubProtocols []p2p.Protocol @@ -91,6 +97,13 @@ type ProtocolManager struct { quitSync chan struct{} noMorePeers chan struct{} + // channels for notarySetLoop + newNotarySetCh chan newNotarySetEvent + newRoundCh chan uint64 + newNotaryNodeInfoCh chan *notaryNodeInfo + + srvr p2pServer + // wait group is used for graceful shutdowns during downloading // and processing wg sync.WaitGroup @@ -101,17 +114,22 @@ type ProtocolManager struct { func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkID uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ - networkID: networkID, - eventMux: mux, - txpool: txpool, - blockchain: blockchain, - chainconfig: config, - peers: newPeerSet(), - newPeerCh: make(chan *peer), - noMorePeers: make(chan struct{}), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), + networkID: networkID, + eventMux: mux, + txpool: txpool, + blockchain: blockchain, + chainconfig: config, + peers: newPeerSet(), + newPeerCh: make(chan *peer), + noMorePeers: make(chan struct{}), + txsyncCh: make(chan *txsync), + quitSync: make(chan struct{}), + newNotarySetCh: make(chan newNotarySetEvent), + newRoundCh: make(chan uint64), + newNotaryNodeInfoCh: make(chan *notaryNodeInfo), } + manager.notarySetManager = newNotarySetManager(manager.newNotaryNodeInfoCh) + // Figure out whether to allow fast sync or not if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 { log.Warn("Blockchain not empty, fast sync disabled") @@ -195,8 +213,9 @@ func (pm *ProtocolManager) removePeer(id string) { } } -func (pm *ProtocolManager) Start(maxPeers int) { +func (pm *ProtocolManager) Start(srvr p2pServer, maxPeers int) { pm.maxPeers = maxPeers + pm.srvr = srvr // broadcast transactions pm.txsCh = make(chan core.NewTxsEvent, txChanSize) @@ -207,6 +226,9 @@ func (pm *ProtocolManager) Start(maxPeers int) { pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{}) go pm.minedBroadcastLoop() + // run the notary set loop + go pm.notarySetLoop() + // start sync handlers go pm.syncer() go pm.txsyncLoop() @@ -677,6 +699,18 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } pm.txpool.AddRemotes(txs) + case msg.Code == NotaryNodeInfoMsg: + var info notaryNodeInfo + if err := msg.Decode(&info); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + + // TODO(sonic): validate signature + p.MarkNotaryNodeInfo(info.Hash()) + // Broadcast to peers + pm.BroadcastNotaryNodeInfo(&info) + + pm.notarySetManager.TryAddInfo(&info) default: return errResp(ErrInvalidMsgCode, "%v", msg.Code) } @@ -735,6 +769,14 @@ func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) { } } +// BroadcastNotaryNodeInfo will propagate notary node info to its peers. +func (pm *ProtocolManager) BroadcastNotaryNodeInfo(info *notaryNodeInfo) { + peers := pm.peers.PeersWithoutNotaryNodeInfo(info.Hash()) + for _, peer := range peers { + peer.AsyncSendNotaryNodeInfo(info) + } +} + // Mined broadcast loop func (pm *ProtocolManager) minedBroadcastLoop() { // automatically stops if unsubscribe @@ -759,6 +801,100 @@ func (pm *ProtocolManager) txBroadcastLoop() { } } +// a loop keep building and maintaining peers in notary set. +func (pm *ProtocolManager) notarySetLoop() { + // TODO: + // * pm need to know current round, and whether we are in notary set. + // * (done) able to handle node info in the near future round. + // * check peer limit. + // * revisit channels (newNotarySetCh, newRoundCh, newNotaryNodeInfoCh) (buffered ?) + + // If we are in notary set and we are synced with network. + // events: + // * new notary set is determined, and we are in notary set. + // (TBD: subscribe the event or polling govornance) + // - advance round + // - build new notary set + // - remove old notary set, remove old notary set from static + + // * current round node info changed. + + self := pm.srvr.Self() + var round uint64 + + for { + select { + case event := <-pm.newNotarySetCh: + // new notary set is determined. + if _, ok := event.Set[self.ID.String()]; ok { + // initialize the new notary set and add it to pm.notarySetManager + s := newNotarySet(event.Round, event.Set) + pm.notarySetManager.Register(event.Round, s) + + // TODO: handle signature + pm.BroadcastNotaryNodeInfo(¬aryNodeInfo{ + ID: self.ID, + IP: self.IP, + UDP: self.UDP, + TCP: self.TCP, + Round: event.Round, + Timestamp: time.Now().Unix(), + }) + } + + case r := <-pm.newRoundCh: + // move to new round. + round = r + + // TODO: revisit this to make sure how many previous round's + // notary set we need to keep. + + // rmove notary set before current round, they are useless + for _, s := range pm.notarySetManager.Before(round) { + pm.removeNotarySetFromStatic(s) + } + + // prepare notary set connections for new round. + if pm.isInNotarySet(round + 1) { + // we assume the notary set for round + 1 is added to + // pm.notarySetManager before this step. + if n, ok := pm.notarySetManager.Round(round + 1); ok { + pm.addNotarySetToStatic(n) + } + } + + case <-pm.newNotaryNodeInfoCh: + // some node info in "current round" is updated. + // try to connect them by the new info. + if pm.isInNotarySet(round) { + if n, ok := pm.notarySetManager.Round(round); ok { + pm.addNotarySetToStatic(n) + } + } + + case <-pm.quitSync: + return + } + } +} + +func (pm *ProtocolManager) isInNotarySet(r uint64) bool { + return false +} + +func (pm *ProtocolManager) addNotarySetToStatic(n *notarySet) { + for _, node := range n.NodesToAdd() { + pm.srvr.AddNotaryPeer(node) + n.MarkAdded(node.ID.String()) + } +} + +func (pm *ProtocolManager) removeNotarySetFromStatic(n *notarySet) { + for _, node := range n.Nodes() { + pm.srvr.RemoveNotaryPeer(node) + } +} + // NodeInfo represents a short summary of the Ethereum sub-protocol metadata // known about the host peer. type NodeInfo struct { diff --git a/dex/helper_test.go b/dex/helper_test.go index 13d2bbaec..8836b31da 100644 --- a/dex/helper_test.go +++ b/dex/helper_test.go @@ -46,6 +46,28 @@ var ( testBank = crypto.PubkeyToAddress(testBankKey.PublicKey) ) +// testP2PServer is a fake, helper p2p server for testing purposes. +type testP2PServer struct { + added chan *discover.Node + removed chan *discover.Node +} + +func (s *testP2PServer) Self() *discover.Node { + return &discover.Node{} +} + +func (s *testP2PServer) AddNotaryPeer(node *discover.Node) { + if s.added != nil { + s.added <- node + } +} + +func (s *testP2PServer) RemoveNotaryPeer(node *discover.Node) { + if s.removed != nil { + s.removed <- node + } +} + // newTestProtocolManager creates a new protocol manager for testing purposes, // with the given number of blocks already known, and potential notification // channels for different events. @@ -70,7 +92,7 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func if err != nil { return nil, nil, err } - pm.Start(1000) + pm.Start(&testP2PServer{}, 1000) return pm, db, nil } diff --git a/dex/protocol.go b/dex/protocol.go index 6452d854a..8aa16db2f 100644 --- a/dex/protocol.go +++ b/dex/protocol.go @@ -22,12 +22,13 @@ import ( "math/big" "net" - "github.com/dexon-foundation/dexon/crypto/sha3" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" ) @@ -113,6 +114,14 @@ type txPool interface { SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription } +type p2pServer interface { + Self() *enode.Node + + AddNotaryPeer(*discover.Node) + + RemoveNotaryPeer(*discover.Node) +} + // statusData is the network packet for the status message. type statusData struct { ProtocolVersion uint32 -- cgit v1.2.3