From 228f1a1875007a58db468318afd5203fe070afd8 Mon Sep 17 00:00:00 2001 From: DinhLN Date: Mon, 22 Oct 2018 15:34:02 +0700 Subject: [PATCH 01/10] Fixed unit test for new randomize smc. --- common/types.go | 2 +- common/types_test.go | 2 +- contracts/randomize/randomize_test.go | 112 +++++++++++++++----------- 3 files changed, 66 insertions(+), 50 deletions(-) diff --git a/common/types.go b/common/types.go index 61949bda1a..8877b5bd1a 100644 --- a/common/types.go +++ b/common/types.go @@ -246,7 +246,7 @@ func (a UnprefixedAddress) MarshalText() ([]byte, error) { // Extract validators from byte array. func RemoveItemFromArray(array []Address, items []Address) []Address { - if items == nil || len(items) == 0 { + if items == nil { return array } for i, value := range array { diff --git a/common/types_test.go b/common/types_test.go index 116cf3d8b7..4b01bc275d 100644 --- a/common/types_test.go +++ b/common/types_test.go @@ -155,6 +155,6 @@ func TestRemoveItemInArray(t *testing.T) { remove := []Address{HexToAddress("0x0000000"), HexToAddress("0x0000004"), HexToAddress("0x0000003")} array = RemoveItemFromArray(array, remove) if len(array) != 2 { - t.Error("fail remove item from array addres ") + t.Error("fail remove item from array address") } } diff --git a/contracts/randomize/randomize_test.go b/contracts/randomize/randomize_test.go index cc129491cc..bf77223813 100644 --- a/contracts/randomize/randomize_test.go +++ b/contracts/randomize/randomize_test.go @@ -85,56 +85,72 @@ func TestSendTxRandomizeSecretAndOpening(t *testing.T) { } backend.Commit() - nonce := uint64(1) randomizeKeyValue := contracts.RandStringByte(32) - tx, err := contracts.BuildTxSecretRandomize(nonce, randomizeAddr, epocNumber, randomizeKeyValue) - if err != nil { - t.Fatalf("Can't create tx randomize secret: %v", err) - } - tx, err = types.SignTx(tx, signer, acc1Key) - if err != nil { - t.Fatalf("Can't sign tx randomize secret: %v", err) - } - err = backend.SendTransaction(ctx, tx) - if err != nil { - t.Fatalf("Can't send tx for create randomize secret: %v", err) - } - backend.Commit() - // Increment nonce. - nonce++ - // Set opening. - tx, err = contracts.BuildTxOpeningRandomize(nonce, randomizeAddr, randomizeKeyValue) - if err != nil { - t.Fatalf("Can't create tx randomize opening: %v", err) - } - tx, err = types.SignTx(tx, signer, acc1Key) - if err != nil { - t.Fatalf("Can't sign tx randomize opening: %v", err) - } + for i := 1; i <= 900; i++ { + nonce := uint64(i) + switch i { + case 800: + tx, err := contracts.BuildTxSecretRandomize(nonce, randomizeAddr, epocNumber, randomizeKeyValue) + if err != nil { + t.Fatalf("Can't create tx randomize secret: %v", err) + } + tx, err = types.SignTx(tx, signer, acc1Key) + if err != nil { + t.Fatalf("Can't sign tx randomize secret: %v", err) + } + err = backend.SendTransaction(ctx, tx) + if err != nil { + t.Fatalf("Can't send tx for create randomize secret: %v", err) + } + break + case 850: + // Set opening. + tx, err := contracts.BuildTxOpeningRandomize(nonce, randomizeAddr, randomizeKeyValue) + if err != nil { + t.Fatalf("Can't create tx randomize opening: %v", err) + } + tx, err = types.SignTx(tx, signer, acc1Key) + if err != nil { + t.Fatalf("Can't sign tx randomize opening: %v", err) + } + err = backend.SendTransaction(ctx, tx) + if err != nil { + t.Fatalf("Can't send tx for create randomize opening: %v", err) + } + break - err = backend.SendTransaction(ctx, tx) - if err != nil { - t.Fatalf("Can't send tx for create randomize opening: %v", err) - } - backend.Commit() - - // Get randomize secret from SC. - secrets, err := randomizeContract.GetSecret(acc1Addr) - if err != nil { - t.Error("Fail get secrets from randomize", err) - } - if len(secrets) <= 0 { - t.Error("Empty get secrets from SC", err) - } - // Decrypt randomize from SC. - opening, err := randomizeContract.GetOpening(acc1Addr) - if err != nil { - t.Fatalf("Can't get secret from SC: %v", err) - } - randomize, err := contracts.DecryptRandomizeFromSecretsAndOpening(secrets, opening) - t.Log("randomize", randomize) - if err != nil { - t.Error("Can't decrypt secret and opening", err) + case 900: + // Get randomize secret from SC. + secrets, err := randomizeContract.GetSecret(acc1Addr) + if err != nil { + t.Error("Fail get secrets from randomize", err) + } + if len(secrets) <= 0 { + t.Error("Empty get secrets from SC", err) + } + // Decrypt randomize from SC. + opening, err := randomizeContract.GetOpening(acc1Addr) + if err != nil { + t.Fatalf("Can't get secret from SC: %v", err) + } + randomize, err := contracts.DecryptRandomizeFromSecretsAndOpening(secrets, opening) + t.Log("randomize", randomize) + if err != nil { + t.Error("Can't decrypt secret and opening", err) + } + break + default: + tx, err := types.SignTx(types.NewTransaction(nonce, common.Address{}, new(big.Int), 21000, new(big.Int), nil), signer, acc1Key) + if err != nil { + t.Fatalf("Can't sign tx randomize: %v", err) + } + err = backend.SendTransaction(ctx, tx) + if err != nil { + t.Fatalf("Can't send tx for create randomize: %v", err) + } + break + } + backend.Commit() } } From 3a49a4af4fb9bce9d8299463629eefa9469d16ba Mon Sep 17 00:00:00 2001 From: DinhLN Date: Thu, 4 Oct 2018 15:15:32 +0700 Subject: [PATCH 02/10] Add penalty feature for prevent signer without sign in epoc make slow down chain. Fixed duplicate nonce value for randomize and sign txs Fixed prevent signers penaltied return in 4 epocs sequent. --- core/blockchain.go | 10 +++++----- eth/backend.go | 1 - 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index efd6c0758e..8b61d18dcd 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -690,7 +690,7 @@ func (bc *BlockChain) procFutureBlocks() { // Insert one by one as chain insertion needs contiguous ancestry between blocks for i := range blocks { - bc.InsertChain(blocks[i : i+1]) + bc.InsertChain(blocks[i: i+1]) } } } @@ -699,9 +699,9 @@ func (bc *BlockChain) procFutureBlocks() { type WriteStatus byte const ( - NonStatTy WriteStatus = iota - CanonStatTy - SideStatTy + NonStatTy WriteStatus = iota + CanonStatTy + SideStatTy ) // Rollback is designed to remove a chain of links from the database that aren't @@ -1243,7 +1243,7 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor if index == len(chain)-1 || elapsed >= statsReportLimit { var ( end = chain[index] - txs = countTransactions(chain[st.lastIndex : index+1]) + txs = countTransactions(chain[st.lastIndex: index+1]) ) context := []interface{}{ "blocks", st.processed, "txs", txs, "mgas", float64(st.usedGas) / 1000000, diff --git a/eth/backend.go b/eth/backend.go index 6299a7017f..c67ed1bad6 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -282,7 +282,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } return penSigners, nil } - return []common.Address{}, nil } From 4c635b57e1ac6056090ebd02db365287527a3463 Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Mon, 1 Oct 2018 16:56:26 +0700 Subject: [PATCH 03/10] add a pair connections with each peer --- eth/handler.go | 52 ++++++++++++++++-------------- eth/peer.go | 20 ++++++++++-- p2p/dial.go | 23 +++++++++++-- p2p/dial_test.go | 44 ++++++++++++++++--------- p2p/peer.go | 10 ++++-- p2p/peer_error.go | 4 +++ p2p/rlpx.go | 1 + p2p/server.go | 17 ++++++++-- p2p/server_test.go | 16 ++++++++- p2p/simulations/adapters/inproc.go | 16 ++++++--- p2p/simulations/http_test.go | 2 +- 11 files changed, 146 insertions(+), 59 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index d46e7689f3..0807f5b1ec 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -271,38 +271,40 @@ func (pm *ProtocolManager) handle(p *peer) error { rw.Init(p.version) } // Register the peer locally - if err := pm.peers.Register(p); err != nil { + err := pm.peers.Register(p) + if err != nil && err != p2p.ErrAddPairPeer { p.Log().Error("Ethereum peer registration failed", "err", err) return err } defer pm.removePeer(p.id) - - // Register the peer in the downloader. If the downloader considers it banned, we disconnect - if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil { - return err - } - // Propagate existing transactions. new transactions appearing - // after this will be sent via broadcasts. - pm.syncTransactions(p) - - // If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork - if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil { - // Request the peer's DAO fork header for extra-data validation - if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil { + if err != p2p.ErrAddPairPeer { + // Register the peer in the downloader. If the downloader considers it banned, we disconnect + if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil { return err } - // Start a timer to disconnect if the peer doesn't reply in time - p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() { - p.Log().Debug("Timed out DAO fork-check, dropping") - pm.removePeer(p.id) - }) - // Make sure it's cleaned up if the peer dies off - defer func() { - if p.forkDrop != nil { - p.forkDrop.Stop() - p.forkDrop = nil + // Propagate existing transactions. new transactions appearing + // after this will be sent via broadcasts. + pm.syncTransactions(p) + + // If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork + if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil { + // Request the peer's DAO fork header for extra-data validation + if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil { + return err } - }() + // Start a timer to disconnect if the peer doesn't reply in time + p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() { + p.Log().Debug("Timed out DAO fork-check, dropping") + pm.removePeer(p.id) + }) + // Make sure it's cleaned up if the peer dies off + defer func() { + if p.forkDrop != nil { + p.forkDrop.Stop() + p.forkDrop = nil + } + }() + } } // main loop. handle incoming messages. for { diff --git a/eth/peer.go b/eth/peer.go index 42ead53965..3201649751 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -25,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" "gopkg.in/fatih/set.v0" @@ -65,6 +66,7 @@ type peer struct { knownTxs *set.Set // Set of transaction hashes known to be known by this peer knownBlocks *set.Set // Set of block hashes known to be known by this peer + pairRw p2p.MsgReadWriter } func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { @@ -156,7 +158,13 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error // SendNewBlock propagates an entire block to a remote peer. func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { p.knownBlocks.Add(block.Hash()) - return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) + if p.pairRw != nil { + log.Trace("p2p SendNewBlock with pairRw", "p", p, "number", block.NumberU64()) + return p2p.Send(p.pairRw, NewBlockMsg, []interface{}{block, td}) + } else { + return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) + } + } // SendBlockHeaders sends a batch of block headers to the remote peer. @@ -321,8 +329,14 @@ func (ps *peerSet) Register(p *peer) error { if ps.closed { return errClosed } - if _, ok := ps.peers[p.id]; ok { - return errAlreadyRegistered + if exitPeer, ok := ps.peers[p.id]; ok { + if exitPeer.pairRw != nil { + return errAlreadyRegistered + } + exitPeer.PairPeer = p.Peer + exitPeer.pairRw = p.rw + p.PairPeer = exitPeer.Peer + return p2p.ErrAddPairPeer } ps.peers[p.id] = p return nil diff --git a/p2p/dial.go b/p2p/dial.go index d8feceb9f3..e87aaaa84c 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -266,7 +266,10 @@ func (s *dialstate) checkDial(n *discover.Node, peers map[discover.NodeID]*Peer) case dialing: return errAlreadyDialing case peers[n.ID] != nil: - return errAlreadyConnected + exitPeer := peers[n.ID] + if exitPeer.PairPeer != nil { + return errAlreadyConnected + } case s.ntab != nil && n.ID == s.ntab.Self().ID: return errSelf case s.netrestrict != nil && !s.netrestrict.Contains(n.IP): @@ -300,10 +303,26 @@ func (t *dialTask) Do(srv *Server) { // Try resolving the ID of static nodes if dialing failed. if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 { if t.resolve(srv) { - t.dial(srv, t.dest) + err = t.dial(srv, t.dest) } } } + if err == nil { + err = t.dial(srv, t.dest) + if err != nil { + // Try resolving the ID of static nodes if dialing failed. + if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 { + if t.resolve(srv) { + err = t.dial(srv, t.dest) + } + } + } + if err == nil { + log.Trace("Dial pair connection sucess", "task", t.dest) + } else { + log.Trace("Dial pair connection error", "task", t.dest, "err", err) + } + } } // resolve attempts to find the current endpoint for the destination diff --git a/p2p/dial_test.go b/p2p/dial_test.go index 2a7941fc65..bcd1b56c88 100644 --- a/p2p/dial_test.go +++ b/p2p/dial_test.go @@ -116,9 +116,9 @@ func TestDialStateDynDial(t *testing.T) { }}, }, new: []task{ + &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(2)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(3)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}}, - &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}}, }, }, // Some of the dials complete but no new ones are launched yet because @@ -164,9 +164,7 @@ func TestDialStateDynDial(t *testing.T) { {rw: &conn{flags: dynDialedConn, id: uintID(4)}}, {rw: &conn{flags: dynDialedConn, id: uintID(5)}}, }, - new: []task{ - &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(6)}}, - }, + new: []task{}, }, // More peers (3,4) drop off and dial for ID 6 completes. // The last query result from the discovery lookup is reused @@ -181,8 +179,8 @@ func TestDialStateDynDial(t *testing.T) { &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(6)}}, }, new: []task{ + &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(7)}}, - &discoverTask{}, }, }, // Peer 7 is connected, but there still aren't enough dynamic peers @@ -212,7 +210,7 @@ func TestDialStateDynDial(t *testing.T) { &discoverTask{}, }, new: []task{ - &discoverTask{}, + &waitExpireTask{Duration: 14 * time.Second}, }, }, }, @@ -302,6 +300,9 @@ func TestDialStateDynDialBootnode(t *testing.T) { &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}}, }, + new: []task{ + &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}}, + }, }, }, }) @@ -351,10 +352,11 @@ func TestDialStateDynDialFromTable(t *testing.T) { }}, }, new: []task{ + &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(1)}}, + &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(2)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(10)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(11)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(12)}}, - &discoverTask{}, }, }, // Dialing nodes 3,4,5 fails. The dials from the lookup succeed. @@ -374,6 +376,9 @@ func TestDialStateDynDialFromTable(t *testing.T) { &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(11)}}, &dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(12)}}, }, + new: []task{ + &discoverTask{}, + }, }, // Waiting for expiry. No waitExpireTask is launched because the // discovery query is still running. @@ -453,6 +458,8 @@ func TestDialStateStaticDial(t *testing.T) { {rw: &conn{flags: dynDialedConn, id: uintID(2)}}, }, new: []task{ + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}}, &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}}, &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}}, @@ -466,6 +473,9 @@ func TestDialStateStaticDial(t *testing.T) { {rw: &conn{flags: dynDialedConn, id: uintID(2)}}, {rw: &conn{flags: staticDialedConn, id: uintID(3)}}, }, + new: []task{ + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}}, + }, done: []task{ &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}}, }, @@ -485,7 +495,8 @@ func TestDialStateStaticDial(t *testing.T) { &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}}, }, new: []task{ - &waitExpireTask{Duration: 14 * time.Second}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}}, }, }, // Wait a round for dial history to expire, no new tasks should spawn. @@ -506,10 +517,7 @@ func TestDialStateStaticDial(t *testing.T) { {rw: &conn{flags: staticDialedConn, id: uintID(3)}}, {rw: &conn{flags: staticDialedConn, id: uintID(5)}}, }, - new: []task{ - &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, - &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}}, - }, + new: []task{}, }, }, }) @@ -542,7 +550,8 @@ func TestDialStaticAfterReset(t *testing.T) { &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, }, new: []task{ - &waitExpireTask{Duration: 30 * time.Second}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, }, }, } @@ -554,7 +563,9 @@ func TestDialStaticAfterReset(t *testing.T) { for _, n := range wantStatic { dTest.init.removeStatic(n) dTest.init.addStatic(n) + delete(dTest.init.dialing, n.ID) } + // without removing peers they will be considered recently dialed runDialTest(t, dTest) } @@ -591,6 +602,10 @@ func TestDialStateCache(t *testing.T) { &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}}, &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, }, + new: []task{ + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}}, + &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}}, + }, }, // A salvage task is launched to wait for node 3's history // entry to expire. @@ -602,9 +617,6 @@ func TestDialStateCache(t *testing.T) { done: []task{ &dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}}, }, - new: []task{ - &waitExpireTask{Duration: 14 * time.Second}, - }, }, // Still waiting for node 3's entry to expire in the cache. { diff --git a/p2p/peer.go b/p2p/peer.go index 477d8c2190..cf8b0d6476 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -108,7 +108,8 @@ type Peer struct { disc chan DiscReason // events receives message send / receive events if set - events *event.Feed + events *event.Feed + PairPeer *Peer } // NewPeer returns a peer for testing purposes. @@ -157,7 +158,7 @@ func (p *Peer) Disconnect(reason DiscReason) { // String implements fmt.Stringer. func (p *Peer) String() string { - return fmt.Sprintf("Peer %x %v", p.rw.id[:8], p.RemoteAddr()) + return fmt.Sprintf("Peer %x %v ", p.rw.id[:8], p.RemoteAddr()) } // Inbound returns true if the peer is an inbound connection @@ -225,10 +226,12 @@ loop: break loop } } - close(p.closed) p.rw.close(reason) p.wg.Wait() + if p.PairPeer != nil { + go func() { p.PairPeer.Disconnect(DiscPairPeerStop) }() + } return remoteRequested, err } @@ -345,6 +348,7 @@ func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error) rw = newMsgEventer(rw, p.events, p.ID(), proto.Name) } p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version)) + go func() { err := proto.Run(p, rw) if err == nil { diff --git a/p2p/peer_error.go b/p2p/peer_error.go index a1cddb707b..544d42315b 100644 --- a/p2p/peer_error.go +++ b/p2p/peer_error.go @@ -54,6 +54,8 @@ func (self *peerError) Error() string { var errProtocolReturned = errors.New("protocol returned") +var ErrAddPairPeer = errors.New("add a pair peer") + type DiscReason uint const ( @@ -69,6 +71,7 @@ const ( DiscUnexpectedIdentity DiscSelf DiscReadTimeout + DiscPairPeerStop DiscSubprotocolError = 0x10 ) @@ -85,6 +88,7 @@ var discReasonToString = [...]string{ DiscUnexpectedIdentity: "unexpected identity", DiscSelf: "connected to self", DiscReadTimeout: "read timeout", + DiscPairPeerStop: "pair peer connection stop", DiscSubprotocolError: "subprotocol error", } diff --git a/p2p/rlpx.go b/p2p/rlpx.go index a320e81e7c..6967de9bff 100644 --- a/p2p/rlpx.go +++ b/p2p/rlpx.go @@ -122,6 +122,7 @@ func (t *rlpx) close(err error) { } func (t *rlpx) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err error) { + // Writing our handshake happens concurrently, we prefer // returning the handshake read error. If the remote side // disconnects us early with a valid reason, we should return it diff --git a/p2p/server.go b/p2p/server.go index c41d1dc156..b81daabec3 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -286,6 +286,7 @@ func (srv *Server) PeerCount() int { // server is shut down. If the connection fails for any reason, the server will // attempt to reconnect the peer. func (srv *Server) AddPeer(node *discover.Node) { + select { case srv.addstatic <- node: case <-srv.quit: @@ -642,9 +643,15 @@ running: p.events = &srv.peerFeed } name := truncateName(c.name) - srv.log.Debug("Adding p2p peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1) + go srv.runPeer(p) - peers[c.id] = p + if peers[c.id] != nil { + peers[c.id].PairPeer = p + srv.log.Debug("Adding p2p pair peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1) + } else { + peers[c.id] = p + srv.log.Debug("Adding p2p peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1) + } if p.Inbound() { inboundCount++ } @@ -708,7 +715,11 @@ func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, inboundCo case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns(): return DiscTooManyPeers case peers[c.id] != nil: - return DiscAlreadyConnected + exitPeer := peers[c.id] + if exitPeer.PairPeer != nil { + return DiscAlreadyConnected + } + return nil case c.id == srv.Self().ID: return DiscSelf default: diff --git a/p2p/server_test.go b/p2p/server_test.go index 10c36528eb..c3af801d74 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -153,7 +153,6 @@ func TestServerDial(t *testing.T) { select { case conn := <-accepted: defer conn.Close() - select { case peer := <-connected: if peer.ID() != remid { @@ -174,6 +173,21 @@ func TestServerDial(t *testing.T) { t.Error("server did not launch peer within one second") } + select { + case peer := <-connected: + if peer.ID() != remid { + t.Errorf("peer has wrong id") + } + if peer.Name() != "test" { + t.Errorf("peer has wrong name") + } + if peer.RemoteAddr().String() != conn.LocalAddr().String() { + t.Errorf("peer started with wrong conn: got %v, want %v", + peer.RemoteAddr(), conn.LocalAddr()) + } + case <-time.After(1 * time.Second): + t.Error("server did not launch peer within one second") + } case <-time.After(1 * time.Second): t.Error("server did not connect within one second") } diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 48d7c17301..cb333f797e 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -91,11 +91,12 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { } simNode := &SimNode{ - ID: id, - config: config, - node: n, - adapter: s, - running: make(map[string]node.Service), + ID: id, + config: config, + node: n, + adapter: s, + running: make(map[string]node.Service), + connected: make(map[discover.NodeID]bool), } s.nodes[id] = simNode return simNode, nil @@ -108,12 +109,16 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) { if !ok { return nil, fmt.Errorf("unknown node: %s", dest.ID) } + if node.connected[dest.ID] { + return nil, fmt.Errorf("dialed node: %s", dest.ID) + } srv := node.Server() if srv == nil { return nil, fmt.Errorf("node not running: %s", dest.ID) } pipe1, pipe2 := net.Pipe() go srv.SetupConn(pipe1, 0, nil) + node.connected[dest.ID] = true return pipe2, nil } @@ -151,6 +156,7 @@ type SimNode struct { running map[string]node.Service client *rpc.Client registerOnce sync.Once + connected map[discover.NodeID]bool } // Addr returns the node's discovery address diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index 677a8fb147..e87550d559 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -179,7 +179,7 @@ func (t *testService) RunDum(p *p2p.Peer, rw p2p.MsgReadWriter) error { // close the dumReady channel so that other protocols can run close(peer.dumReady) - + // block until the peer is dropped for { _, err := rw.ReadMsg() From bcfb59fe14250902c6aed4324e6e2fd65981410c Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Fri, 5 Oct 2018 16:02:23 +0700 Subject: [PATCH 04/10] Broadcast special Tx through pairRW --- core/blockchain.go | 10 ++-- core/tx_list.go | 21 ++++----- core/tx_pool.go | 84 ++++++++++++++++++++++++++++------ core/types/transaction.go | 40 +++++++++++++--- core/types/transaction_test.go | 2 +- eth/handler.go | 30 ++++++++++++ eth/helper_test.go | 4 ++ eth/peer.go | 11 ++++- eth/protocol.go | 1 + internal/ethapi/api.go | 4 +- miner/worker.go | 53 ++++++++++++++++++--- p2p/simulations/http_test.go | 2 +- 12 files changed, 214 insertions(+), 48 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 8b61d18dcd..efd6c0758e 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -690,7 +690,7 @@ func (bc *BlockChain) procFutureBlocks() { // Insert one by one as chain insertion needs contiguous ancestry between blocks for i := range blocks { - bc.InsertChain(blocks[i: i+1]) + bc.InsertChain(blocks[i : i+1]) } } } @@ -699,9 +699,9 @@ func (bc *BlockChain) procFutureBlocks() { type WriteStatus byte const ( - NonStatTy WriteStatus = iota - CanonStatTy - SideStatTy + NonStatTy WriteStatus = iota + CanonStatTy + SideStatTy ) // Rollback is designed to remove a chain of links from the database that aren't @@ -1243,7 +1243,7 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor if index == len(chain)-1 || elapsed >= statsReportLimit { var ( end = chain[index] - txs = countTransactions(chain[st.lastIndex: index+1]) + txs = countTransactions(chain[st.lastIndex : index+1]) ) context := []interface{}{ "blocks", st.processed, "txs", txs, "mgas", float64(st.usedGas) / 1000000, diff --git a/core/tx_list.go b/core/tx_list.go index 12134cc162..80bee54825 100644 --- a/core/tx_list.go +++ b/core/tx_list.go @@ -251,19 +251,18 @@ func (l *txList) Overlaps(tx *types.Transaction) bool { func (l *txList) Add(tx *types.Transaction, priceBump uint64) (bool, *types.Transaction) { // If there's an older better transaction, abort old := l.txs.Get(tx.Nonce()) - - if (tx.To() != nil && tx.To().String() != common.RandomizeSMC) || tx.To() == nil { - if old != nil { - threshold := new(big.Int).Div(new(big.Int).Mul(old.GasPrice(), big.NewInt(100+int64(priceBump))), big.NewInt(100)) - // Have to ensure that the new gas price is higher than the old gas - // price as well as checking the percentage threshold to ensure that - // this is accurate for low (Wei-level) gas price replacements - if old.GasPrice().Cmp(tx.GasPrice()) >= 0 || threshold.Cmp(tx.GasPrice()) > 0 { - return false, nil - } + if old != nil && old.IsSpecialTransaction() { + return false, nil + } + if old != nil { + threshold := new(big.Int).Div(new(big.Int).Mul(old.GasPrice(), big.NewInt(100+int64(priceBump))), big.NewInt(100)) + // Have to ensure that the new gas price is higher than the old gas + // price as well as checking the percentage threshold to ensure that + // this is accurate for low (Wei-level) gas price replacements + if old.GasPrice().Cmp(tx.GasPrice()) >= 0 || threshold.Cmp(tx.GasPrice()) > 0 { + return false, nil } } - // Otherwise overwrite the old transaction with the current one l.txs.Put(tx) if cost := tx.Cost(); l.costcap.Cmp(cost) < 0 { diff --git a/core/tx_pool.go b/core/tx_pool.go index 80efaefe78..4ef3a7b674 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -80,6 +80,8 @@ var ( ErrOversizedData = errors.New("oversized data") ErrZeroGasPrice = errors.New("zero gas price") + + ErrDuplicateSpecialTransaction = errors.New("duplicate a specail transaction") ) var ( @@ -186,16 +188,17 @@ func (config *TxPoolConfig) sanitize() TxPoolConfig { // current state) and future transactions. Transactions move between those // two states over time as they are received and processed. type TxPool struct { - config TxPoolConfig - chainconfig *params.ChainConfig - chain blockChain - gasPrice *big.Int - txFeed event.Feed - scope event.SubscriptionScope - chainHeadCh chan ChainHeadEvent - chainHeadSub event.Subscription - signer types.Signer - mu sync.RWMutex + config TxPoolConfig + chainconfig *params.ChainConfig + chain blockChain + gasPrice *big.Int + txFeed event.Feed + specialTxFeed event.Feed + scope event.SubscriptionScope + chainHeadCh chan ChainHeadEvent + chainHeadSub event.Subscription + signer types.Signer + mu sync.RWMutex currentState *state.StateDB // Current state in the blockchain head pendingState *state.ManagedState // Pending state tracking virtual nonces @@ -454,6 +457,10 @@ func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription return pool.scope.Track(pool.txFeed.Subscribe(ch)) } +func (pool *TxPool) SubscribeSpecialTxPreEvent(ch chan<- TxPreEvent) event.Subscription { + return pool.scope.Track(pool.specialTxFeed.Subscribe(ch)) +} + // GasPrice returns the current gas price enforced by the transaction pool. func (pool *TxPool) GasPrice() *big.Int { pool.mu.RLock() @@ -576,7 +583,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { } // Drop non-local transactions under our own minimal accepted gas price local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network - if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { + if !local && tx.To() != nil && !tx.IsSpecialTransaction() && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { return ErrUnderpriced } // Ensure the transaction adheres to nonce ordering @@ -589,7 +596,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { return ErrInsufficientFunds } - if tx.To() != nil && tx.To().String() != common.BlockSigners && tx.To().String() != common.RandomizeSMC { + if tx.To() != nil && !tx.IsSpecialTransaction() { intrGas, err := IntrinsicGas(tx.Data(), tx.To() == nil, pool.homestead) if err != nil { return err @@ -646,8 +653,11 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { pool.removeTx(tx.Hash()) } } - // If the transaction is replacing an already pending one, do directly from, _ := types.Sender(pool.signer, tx) // already validated + if tx.IsSpecialTransaction() { + return pool.promoteSpecialTx(from, tx) + } + // If the transaction is replacing an already pending one, do directly if list := pool.pending[from]; list != nil && list.Overlaps(tx) { // Nonce already pending, check if required price bump is met inserted, old := list.Add(tx, pool.config.PriceBump) @@ -763,6 +773,54 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T go pool.txFeed.Send(TxPreEvent{tx}) } +func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction) (bool, error) { + // Try to insert the transaction into the pending queue + if pool.pending[addr] == nil { + pool.pending[addr] = newTxList(true) + } + list := pool.pending[addr] + old := list.txs.Get(tx.Nonce()) + if old != nil && old.IsSpecialTransaction() { + return false, ErrDuplicateSpecialTransaction + } + // Otherwise discard any previous transaction and mark this + if old != nil { + delete(pool.all, old.Hash()) + pool.priced.Removed() + pendingReplaceCounter.Inc(1) + } + list.txs.Put(tx) + if cost := tx.Cost(); list.costcap.Cmp(cost) < 0 { + list.costcap = cost + } + if gas := tx.Gas(); list.gascap < gas { + list.gascap = gas + } + // Failsafe to work around direct pending inserts (tests) + if pool.all[tx.Hash()] == nil { + pool.all[tx.Hash()] = tx + } + // Set the potentially new pending nonce and notify any subsystems of the new tx + pool.beats[addr] = time.Now() + pool.pendingState.SetNonce(addr, tx.Nonce()+1) + broadcastTxs := types.Transactions{} + for i := tx.Nonce() - 1; i > 0; i-- { + before := list.txs.Get(i) + if before == nil || before.IsSpecialTransaction() { + break + } + broadcastTxs = append(broadcastTxs, before) + } + broadcastTxs = append(broadcastTxs, tx) + go func() { + for _, btx := range broadcastTxs { + pool.specialTxFeed.Send(TxPreEvent{btx}) + log.Debug("Pooled new special transaction", "hash", tx.Hash(), "from", addr, "to", tx.To(), "nonce", tx.Nonce()) + } + }() + return true, nil +} + // AddLocal enqueues a single transaction into the pool if it is valid, marking // the sender as a local one in the mean time, ensuring it goes around the local // pricing constraints. diff --git a/core/types/transaction.go b/core/types/transaction.go index 11b06fc561..3ce5ae524c 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -267,6 +267,14 @@ func (tx *Transaction) RawSignatureValues() (*big.Int, *big.Int, *big.Int) { return tx.data.V, tx.data.R, tx.data.S } +func (tx *Transaction) IsSpecialTransaction() bool { + to := "" + if tx.To() != nil { + to = tx.To().String() + } + return to == common.RandomizeSMC || to == common.BlockSigners +} + func (tx *Transaction) String() string { var from, to string if tx.data.V != nil { @@ -395,14 +403,32 @@ type TransactionsByPriceAndNonce struct { // // Note, the input map is reowned so the caller should not interact any more with // if after providing it to the constructor. -func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) *TransactionsByPriceAndNonce { +func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) (*TransactionsByPriceAndNonce, Transactions) { // Initialize a price based heap with the head transactions - heads := make(TxByPrice, 0, len(txs)) + heads := TxByPrice{} + specialTxs := Transactions{} for _, accTxs := range txs { - heads = append(heads, accTxs[0]) - // Ensure the sender address is from the signer - acc, _ := Sender(signer, accTxs[0]) - txs[acc] = accTxs[1:] + var normalTxs Transactions + lastSpecialTx := -1 + for i, tx := range accTxs { + if tx.IsSpecialTransaction() { + lastSpecialTx = i + } + } + if lastSpecialTx >= 0 { + for i := 0; i <= lastSpecialTx; i++ { + specialTxs = append(specialTxs, accTxs[i]) + } + normalTxs = accTxs[lastSpecialTx+1:] + } else { + normalTxs = accTxs + } + if len(normalTxs) > 0 { + acc, _ := Sender(signer, normalTxs[0]) + heads = append(heads, normalTxs[0]) + // Ensure the sender address is from the signer + txs[acc] = normalTxs[1:] + } } heap.Init(&heads) @@ -411,7 +437,7 @@ func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transa txs: txs, heads: heads, signer: signer, - } + }, specialTxs } // Peek returns the next transaction by price. diff --git a/core/types/transaction_test.go b/core/types/transaction_test.go index d1861b14cd..602fe8fe23 100644 --- a/core/types/transaction_test.go +++ b/core/types/transaction_test.go @@ -144,7 +144,7 @@ func TestTransactionPriceNonceSort(t *testing.T) { } } // Sort the transactions and cross check the nonce ordering - txset := NewTransactionsByPriceAndNonce(signer, groups) + txset, _ := NewTransactionsByPriceAndNonce(signer, groups) txs := Transactions{} for tx := txset.Peek(); tx != nil; tx = txset.Peek() { diff --git a/eth/handler.go b/eth/handler.go index 0807f5b1ec..c00a54849a 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -83,6 +83,8 @@ type ProtocolManager struct { eventMux *event.TypeMux txCh chan core.TxPreEvent txSub event.Subscription + specialTxCh chan core.TxPreEvent + specialTxSub event.Subscription minedBlockSub *event.TypeMuxSubscription // channels for fetcher, syncer, txsyncLoop @@ -208,6 +210,10 @@ func (pm *ProtocolManager) Start(maxPeers int) { pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh) go pm.txBroadcastLoop() + pm.specialTxCh = make(chan core.TxPreEvent, txChanSize) + pm.specialTxSub = pm.txpool.SubscribeSpecialTxPreEvent(pm.specialTxCh) + go pm.specialTxBroadcastLoop() + // broadcast mined blocks pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{}) go pm.minedBroadcastLoop() @@ -221,6 +227,7 @@ func (pm *ProtocolManager) Stop() { log.Info("Stopping Ethereum protocol") pm.txSub.Unsubscribe() // quits txBroadcastLoop + pm.specialTxSub.Unsubscribe() // quits specialTxBroadcastLoop pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop // Quit the sync loop. @@ -726,6 +733,16 @@ func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) log.Trace("Broadcast transaction", "hash", hash, "recipients", len(peers)) } +func (pm *ProtocolManager) BroadcastSpecialTx(hash common.Hash, tx *types.Transaction) { + // Broadcast transaction to a batch of peers not knowing about it + peers := pm.peers.PeersWithoutTx(hash) + //FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))] + for _, peer := range peers { + peer.SendSpecialTransactions(tx) + } + log.Debug("Broadcast special transaction", "hash", hash, "recipients", len(peers)) +} + // Mined broadcast loop func (self *ProtocolManager) minedBroadcastLoop() { // automatically stops if unsubscribe @@ -751,6 +768,19 @@ func (self *ProtocolManager) txBroadcastLoop() { } } +func (self *ProtocolManager) specialTxBroadcastLoop() { + for { + select { + case event := <-self.specialTxCh: + self.BroadcastSpecialTx(event.Tx.Hash(), event.Tx) + + // Err() channel will be closed when unsubscribing. + case <-self.specialTxSub.Err(): + return + } + } +} + // NodeInfo represents a short summary of the Ethereum sub-protocol metadata // known about the host peer. type NodeInfo struct { diff --git a/eth/helper_test.go b/eth/helper_test.go index 2b05cea801..ca686df808 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -128,6 +128,10 @@ func (p *testTxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscr return p.txFeed.Subscribe(ch) } +func (p *testTxPool) SubscribeSpecialTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription { + return p.txFeed.Subscribe(ch) +} + // newTestTransaction create a new dummy transaction. func newTestTransaction(from *ecdsa.PrivateKey, nonce uint64, datasize int) *types.Transaction { tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), 100000, big.NewInt(0), make([]byte, datasize)) diff --git a/eth/peer.go b/eth/peer.go index 3201649751..73324b90f2 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -141,6 +141,15 @@ func (p *peer) SendTransactions(txs types.Transactions) error { return p2p.Send(p.rw, TxMsg, txs) } +func (p *peer) SendSpecialTransactions(tx *types.Transaction) error { + p.knownTxs.Add(tx.Hash()) + if p.pairRw != nil { + return p2p.Send(p.pairRw, TxMsg, types.Transactions{tx}) + } else { + return p2p.Send(p.rw, TxMsg, types.Transactions{tx}) + } +} + // SendNewBlockHashes announces the availability of a number of blocks through // a hash notification. func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error { @@ -159,7 +168,7 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { p.knownBlocks.Add(block.Hash()) if p.pairRw != nil { - log.Trace("p2p SendNewBlock with pairRw", "p", p, "number", block.NumberU64()) + log.Trace("p2p Send New Block with pairRw", "p", p, "number", block.NumberU64()) return p2p.Send(p.pairRw, NewBlockMsg, []interface{}{block, td}) } else { return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) diff --git a/eth/protocol.go b/eth/protocol.go index cd7db57f23..f44a01f020 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -106,6 +106,7 @@ type txPool interface { // SubscribeTxPreEvent should return an event subscription of // TxPreEvent and send events to the given channel. SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription + SubscribeSpecialTxPreEvent(chan<- core.TxPreEvent) event.Subscription } // statusData is the network packet for the status message. diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index afe30b141c..7ecdd76787 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1210,8 +1210,8 @@ func (args *SendTxArgs) toTransaction() *types.Transaction { // submitTransaction is a helper function that submits tx to txPool and logs a message. func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (common.Hash, error) { - if tx.To() != nil && tx.To().String() == common.BlockSigners { - return common.Hash{}, errors.New("Dont allow transaction sent to BlockSigners smart contract via API") + if tx.To() != nil && tx.IsSpecialTransaction() { + return common.Hash{}, errors.New("Dont allow transaction sent to BlockSigners & RandomizeSMC smart contract via API") } if err := b.SendTx(ctx, tx); err != nil { return common.Hash{}, err diff --git a/miner/worker.go b/miner/worker.go index 67af1a8ce7..b4286b50aa 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -270,9 +270,9 @@ func (self *worker) update() { self.currentMu.Lock() acc, _ := types.Sender(self.current.signer, ev.Tx) txs := map[common.Address]types.Transactions{acc: {ev.Tx}} - txset := types.NewTransactionsByPriceAndNonce(self.current.signer, txs) + txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs) - self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase) + self.current.commitTransactions(self.mux, txset, specialTxs, self.chain, self.coinbase) self.currentMu.Unlock() } else { // If we're mining, but nothing is being processed, wake on new transactions @@ -280,7 +280,6 @@ func (self *worker) update() { self.commitNewWork() } } - // System stopped case <-self.txSub.Err(): return @@ -552,8 +551,8 @@ func (self *worker) commitNewWork() { log.Error("Failed to fetch pending transactions", "err", err) return } - txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending) - work.commitTransactions(self.mux, txs, self.chain, self.coinbase) + txs, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending) + work.commitTransactions(self.mux, txs, specialTxs, self.chain, self.coinbase) // compute uncles for the new block. var ( @@ -584,7 +583,7 @@ func (self *worker) commitNewWork() { } // We only care about logging if we're actually mining. if atomic.LoadInt32(&self.mining) == 1 { - log.Info("Commit new mining work", "number", work.Block.Number(), "txs", work.tcount, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart))) + log.Info("Commit new mining work", "number", work.Block.Number(), "txs", work.tcount, "special txs", len(specialTxs), "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart))) self.unconfirmed.Shift(work.Block.NumberU64() - 1) } if work.config.Posv != nil { @@ -623,11 +622,51 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error { return nil } -func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) { +func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, specialTxs types.Transactions, bc *core.BlockChain, coinbase common.Address) { gp := new(core.GasPool).AddGas(env.header.GasLimit) var coalescedLogs []*types.Log + for _, tx := range specialTxs { + if gp.Gas() < params.TxGas && tx.Gas() > 0 { + log.Trace("Not enough gas for further transactions", "gp", gp) + break + } + // Error may be ignored here. The error has already been checked + // during transaction acceptance is the transaction pool. + // + // We use the eip155 signer regardless of the current hf. + from, _ := types.Sender(env.signer, tx) + // Check whether the tx is replay protected. If we're not in the EIP155 hf + // phase, start ignoring the sender until we do. + if tx.Protected() && !env.config.IsEIP155(env.header.Number) { + log.Debug("Ignoring reply protected special transaction", "hash", tx.Hash(), "eip155", env.config.EIP155Block) + continue + } + // Start executing the transaction + env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount) + err, logs := env.commitTransaction(tx, bc, coinbase, gp) + switch err { + case core.ErrNonceTooLow: + // New head notification data race between the transaction pool and miner, shift + log.Debug("Skipping special transaction with low nonce", "sender", from, "nonce", tx.Nonce(), "to", tx.To()) + + case core.ErrNonceTooHigh: + // Reorg notification data race between the transaction pool and miner, skip account = + log.Debug("Skipping account with special transaction hight nonce", "sender", from, "nonce", tx.Nonce(), "to", tx.To()) + + case nil: + // Everything ok, collect the logs and shift in the next transaction from the same account + coalescedLogs = append(coalescedLogs, logs...) + env.tcount++ + + default: + // Strange error, discard the transaction and get the next in line (note, the + // nonce-too-high clause will prevent us from executing in vain). + log.Debug("Add Special Transaction failed, account skipped", "hash", tx.Hash(), "sender", from, "nonce", tx.Nonce(), "to", tx.To(), "err", err) + } + } + for { // If we don't have enough gas for any further transactions then we're done if gp.Gas() < params.TxGas { diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index e87550d559..677a8fb147 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -179,7 +179,7 @@ func (t *testService) RunDum(p *p2p.Peer, rw p2p.MsgReadWriter) error { // close the dumReady channel so that other protocols can run close(peer.dumReady) - + // block until the peer is dropped for { _, err := rw.ReadMsg() From a9e181e5a6e612df10d92bf4c2a246997a368f51 Mon Sep 17 00:00:00 2001 From: DinhLN Date: Thu, 4 Oct 2018 15:15:32 +0700 Subject: [PATCH 05/10] Add penalty feature for prevent signer without sign in epoc make slow down chain. Fixed duplicate nonce value for randomize and sign txs Fixed prevent signers penaltied return in 4 epocs sequent. --- core/blockchain.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index efd6c0758e..8b61d18dcd 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -690,7 +690,7 @@ func (bc *BlockChain) procFutureBlocks() { // Insert one by one as chain insertion needs contiguous ancestry between blocks for i := range blocks { - bc.InsertChain(blocks[i : i+1]) + bc.InsertChain(blocks[i: i+1]) } } } @@ -699,9 +699,9 @@ func (bc *BlockChain) procFutureBlocks() { type WriteStatus byte const ( - NonStatTy WriteStatus = iota - CanonStatTy - SideStatTy + NonStatTy WriteStatus = iota + CanonStatTy + SideStatTy ) // Rollback is designed to remove a chain of links from the database that aren't @@ -1243,7 +1243,7 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor if index == len(chain)-1 || elapsed >= statsReportLimit { var ( end = chain[index] - txs = countTransactions(chain[st.lastIndex : index+1]) + txs = countTransactions(chain[st.lastIndex: index+1]) ) context := []interface{}{ "blocks", st.processed, "txs", txs, "mgas", float64(st.usedGas) / 1000000, From 192a9852536691c5c8d4f511f0592573be4e43f5 Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Mon, 1 Oct 2018 16:56:26 +0700 Subject: [PATCH 06/10] add a pair connections with each peer --- eth/peer.go | 2 +- p2p/simulations/http_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/eth/peer.go b/eth/peer.go index 73324b90f2..faf1c16a2b 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -168,7 +168,7 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { p.knownBlocks.Add(block.Hash()) if p.pairRw != nil { - log.Trace("p2p Send New Block with pairRw", "p", p, "number", block.NumberU64()) + log.Trace("p2p send new block to the pairRw connection", "p", p, "number", block.NumberU64()) return p2p.Send(p.pairRw, NewBlockMsg, []interface{}{block, td}) } else { return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index 677a8fb147..e87550d559 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -179,7 +179,7 @@ func (t *testService) RunDum(p *p2p.Peer, rw p2p.MsgReadWriter) error { // close the dumReady channel so that other protocols can run close(peer.dumReady) - + // block until the peer is dropped for { _, err := rw.ReadMsg() From c243f37fc451bc3455bd44f3d0c7f8baea1cd00e Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Fri, 5 Oct 2018 16:02:23 +0700 Subject: [PATCH 07/10] Broadcast special Tx through pairRW --- core/blockchain.go | 10 +++++----- p2p/simulations/http_test.go | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 8b61d18dcd..efd6c0758e 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -690,7 +690,7 @@ func (bc *BlockChain) procFutureBlocks() { // Insert one by one as chain insertion needs contiguous ancestry between blocks for i := range blocks { - bc.InsertChain(blocks[i: i+1]) + bc.InsertChain(blocks[i : i+1]) } } } @@ -699,9 +699,9 @@ func (bc *BlockChain) procFutureBlocks() { type WriteStatus byte const ( - NonStatTy WriteStatus = iota - CanonStatTy - SideStatTy + NonStatTy WriteStatus = iota + CanonStatTy + SideStatTy ) // Rollback is designed to remove a chain of links from the database that aren't @@ -1243,7 +1243,7 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor if index == len(chain)-1 || elapsed >= statsReportLimit { var ( end = chain[index] - txs = countTransactions(chain[st.lastIndex: index+1]) + txs = countTransactions(chain[st.lastIndex : index+1]) ) context := []interface{}{ "blocks", st.processed, "txs", txs, "mgas", float64(st.usedGas) / 1000000, diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index e87550d559..677a8fb147 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -179,7 +179,7 @@ func (t *testService) RunDum(p *p2p.Peer, rw p2p.MsgReadWriter) error { // close the dumReady channel so that other protocols can run close(peer.dumReady) - + // block until the peer is dropped for { _, err := rw.ReadMsg() From 740e6076e53653aa1622a1d74de65c00b5488fde Mon Sep 17 00:00:00 2001 From: Tuna Date: Mon, 22 Oct 2018 11:57:52 +0700 Subject: [PATCH 08/10] fix after reviewing --- core/tx_pool.go | 2 ++ core/types/transaction.go | 9 +++++---- eth/handler.go | 1 + eth/peer.go | 14 +++++++------- miner/worker.go | 1 + p2p/dial.go | 4 ++-- 6 files changed, 18 insertions(+), 13 deletions(-) diff --git a/core/tx_pool.go b/core/tx_pool.go index 4ef3a7b674..d82619e952 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -457,6 +457,8 @@ func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription return pool.scope.Track(pool.txFeed.Subscribe(ch)) } +// SubscribeSpecialTxPreEvent registers a subscription of TxPreEvent and +// starts sending event to the given channel. func (pool *TxPool) SubscribeSpecialTxPreEvent(ch chan<- TxPreEvent) event.Subscription { return pool.scope.Track(pool.specialTxFeed.Subscribe(ch)) } diff --git a/core/types/transaction.go b/core/types/transaction.go index 3ce5ae524c..193447028a 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -268,11 +268,10 @@ func (tx *Transaction) RawSignatureValues() (*big.Int, *big.Int, *big.Int) { } func (tx *Transaction) IsSpecialTransaction() bool { - to := "" - if tx.To() != nil { - to = tx.To().String() + if tx.To() == nil { + return false } - return to == common.RandomizeSMC || to == common.BlockSigners + return tx.To().String() == common.RandomizeSMC || tx.To().String() == common.BlockSigners } func (tx *Transaction) String() string { @@ -403,6 +402,8 @@ type TransactionsByPriceAndNonce struct { // // Note, the input map is reowned so the caller should not interact any more with // if after providing it to the constructor. + +// It also classifies special txs and normal txs func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) (*TransactionsByPriceAndNonce, Transactions) { // Initialize a price based heap with the head transactions heads := TxByPrice{} diff --git a/eth/handler.go b/eth/handler.go index c00a54849a..d2e4ec3211 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -210,6 +210,7 @@ func (pm *ProtocolManager) Start(maxPeers int) { pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh) go pm.txBroadcastLoop() + // broadcast special transactions pm.specialTxCh = make(chan core.TxPreEvent, txChanSize) pm.specialTxSub = pm.txpool.SubscribeSpecialTxPreEvent(pm.specialTxCh) go pm.specialTxBroadcastLoop() diff --git a/eth/peer.go b/eth/peer.go index faf1c16a2b..75813fe93a 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -55,7 +55,8 @@ type peer struct { id string *p2p.Peer - rw p2p.MsgReadWriter + rw p2p.MsgReadWriter + pairRw p2p.MsgReadWriter version int // Protocol version negotiated forkDrop *time.Timer // Timed connection dropper if forks aren't validated in time @@ -66,7 +67,6 @@ type peer struct { knownTxs *set.Set // Set of transaction hashes known to be known by this peer knownBlocks *set.Set // Set of block hashes known to be known by this peer - pairRw p2p.MsgReadWriter } func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { @@ -338,13 +338,13 @@ func (ps *peerSet) Register(p *peer) error { if ps.closed { return errClosed } - if exitPeer, ok := ps.peers[p.id]; ok { - if exitPeer.pairRw != nil { + if existPeer, ok := ps.peers[p.id]; ok { + if existPeer.pairRw != nil { return errAlreadyRegistered } - exitPeer.PairPeer = p.Peer - exitPeer.pairRw = p.rw - p.PairPeer = exitPeer.Peer + existPeer.PairPeer = p.Peer + existPeer.pairRw = p.rw + p.PairPeer = existPeer.Peer return p2p.ErrAddPairPeer } ps.peers[p.id] = p diff --git a/miner/worker.go b/miner/worker.go index b4286b50aa..f421cd5dd6 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -627,6 +627,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB var coalescedLogs []*types.Log + // first priority for special Txs for _, tx := range specialTxs { if gp.Gas() < params.TxGas && tx.Gas() > 0 { log.Trace("Not enough gas for further transactions", "gp", gp) diff --git a/p2p/dial.go b/p2p/dial.go index e87aaaa84c..3a6cff18c1 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -266,8 +266,8 @@ func (s *dialstate) checkDial(n *discover.Node, peers map[discover.NodeID]*Peer) case dialing: return errAlreadyDialing case peers[n.ID] != nil: - exitPeer := peers[n.ID] - if exitPeer.PairPeer != nil { + exitsPeer := peers[n.ID] + if exitsPeer.PairPeer != nil { return errAlreadyConnected } case s.ntab != nil && n.ID == s.ntab.Self().ID: From a7c149d76b275d4e6a125022b58c1f6edd5cc0f4 Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Mon, 8 Oct 2018 14:18:34 +0700 Subject: [PATCH 09/10] verify validators info at checkpoint block --- consensus/posv/api.go | 4 +- consensus/posv/posv.go | 28 ++++++++++---- consensus/posv/snapshot.go | 4 +- core/error.go | 2 + eth/backend.go | 79 +++++++++++++++++++++++++------------- 5 files changed, 78 insertions(+), 39 deletions(-) diff --git a/consensus/posv/api.go b/consensus/posv/api.go index 33c0e77772..fe68560c8a 100644 --- a/consensus/posv/api.go +++ b/consensus/posv/api.go @@ -71,7 +71,7 @@ func (api *API) GetSigners(number *rpc.BlockNumber) ([]common.Address, error) { if err != nil { return nil, err } - return snap.signers(), nil + return snap.GetSigners(), nil } // GetSignersAtHash retrieves the state snapshot at a given block. @@ -84,7 +84,7 @@ func (api *API) GetSignersAtHash(hash common.Hash) ([]common.Address, error) { if err != nil { return nil, err } - return snap.signers(), nil + return snap.GetSigners(), nil } // Proposals returns the current proposals the node tries to uphold and vote on. diff --git a/consensus/posv/posv.go b/consensus/posv/posv.go index 3cb9137878..303568fcc6 100644 --- a/consensus/posv/posv.go +++ b/consensus/posv/posv.go @@ -136,6 +136,8 @@ var ( // on an instant chain (0 second period). It's important to refuse these as the // block reward is zero, so an empty block just bloats the chain... fast. errWaitTransactions = errors.New("waiting for transactions") + + ErrInvalidCheckpointValidators = errors.New("invalid validators list on checkpoint block") ) // SignerFn is a signer callback function to request a hash to be signed by a @@ -213,9 +215,10 @@ type Posv struct { signFn clique.SignerFn // Signer function to authorize hashes with lock sync.RWMutex // Protects the signer fields - HookReward func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error - HookPenalty func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) - HookPrepare func(header *types.Header, signers []common.Address) error + HookReward func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error + HookPenalty func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) + HookPrepare func(header *types.Header, signers []common.Address) error + VerifyValidators func(header *types.Header, signers []common.Address) error } // New creates a Posv proof-of-stake-voting consensus engine with the initial @@ -379,7 +382,7 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types. return errInvalidCheckpointPenalties } } - signers := snap.signers() + signers := snap.GetSigners() signers = common.RemoveItemFromArray(signers, penPenalties) for i := 1; i <= common.LimitPenaltyEpoch; i++ { if number > uint64(i)*c.config.Epoch { @@ -391,6 +394,12 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types. if !bytes.Equal(header.Extra[extraVanity:extraSuffix], byteMasterNodes) { return errInvalidCheckpointSigners } + if c.VerifyValidators != nil { + err := c.VerifyValidators(header, signers) + if err != nil { + return err + } + } } // All basic checks passed, verify the seal and return return c.verifySeal(chain, header, parents) @@ -581,7 +590,7 @@ func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, par mstring = append(mstring, m.String()) } nstring := []string{} - for _, n := range snap.signers() { + for _, n := range snap.GetSigners() { nstring = append(nstring, n.String()) } if _, ok := snap.Signers[signer]; !ok { @@ -656,7 +665,7 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error header.Extra = append(header.Extra, bytes.Repeat([]byte{0x00}, extraVanity-len(header.Extra))...) } header.Extra = header.Extra[:extraVanity] - signers := snap.signers() + signers := snap.GetSigners() if number%c.config.Epoch == 0 { if c.HookPenalty != nil { penSigners, err := c.HookPenalty(chain, number) @@ -698,6 +707,9 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error } if c.HookPrepare != nil { c.HookPrepare(header, signers) + if err != nil { + return err + } } return nil } @@ -710,7 +722,7 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head if err != nil { return err } - currentSigners := snap.signers() + currentSigners := snap.GetSigners() proposedSigners := make(map[common.Address]struct{}) // count all addresses in ms to be masternode for _, m := range ms { @@ -724,7 +736,7 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head } } nm := []string{} - newSigners := snap.signers() + newSigners := snap.GetSigners() for _, n := range newSigners { nm = append(nm, n.String()) } diff --git a/consensus/posv/snapshot.go b/consensus/posv/snapshot.go index 5240a7e3f8..bc75b0d320 100644 --- a/consensus/posv/snapshot.go +++ b/consensus/posv/snapshot.go @@ -286,7 +286,7 @@ func (s *Snapshot) apply(headers []*types.Header) (*Snapshot, error) { } // signers retrieves the list of authorized signers in ascending order. -func (s *Snapshot) signers() []common.Address { +func (s *Snapshot) GetSigners() []common.Address { signers := make([]common.Address, 0, len(s.Signers)) for signer := range s.Signers { signers = append(signers, signer) @@ -303,7 +303,7 @@ func (s *Snapshot) signers() []common.Address { // inturn returns if a signer at a given block height is in-turn or not. func (s *Snapshot) inturn(number uint64, signer common.Address) bool { - signers, offset := s.signers(), 0 + signers, offset := s.GetSigners(), 0 for offset < len(signers) && signers[offset] != signer { offset++ } diff --git a/core/error.go b/core/error.go index d55c7840e6..86b093b151 100644 --- a/core/error.go +++ b/core/error.go @@ -34,4 +34,6 @@ var ( ErrNonceTooHigh = errors.New("nonce too high") ErrNotPoSV = errors.New("Posv not found in config") + + ErrNotFoundM1 = errors.New("list M1 not found ") ) diff --git a/eth/backend.go b/eth/backend.go index c67ed1bad6..f0ed406c0f 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -25,6 +25,7 @@ import ( "sync" "sync/atomic" + "bytes" "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" @@ -211,36 +212,13 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { // Hook will process when preparing block. c.HookPrepare = func(header *types.Header, signers []common.Address) error { - client, err := eth.blockchain.GetClient() - if err != nil { - log.Error("Fail to connect IPC client for penalty.", "error", err) - } number := header.Number.Int64() - // Check m2 exists on chaindb. - // Get secrets and opening at epoc block checkpoint. if number > 0 && number%common.EpocBlockRandomize == 0 { - var candidates []int64 - lenSigners := int64(len(signers)) - - if lenSigners > 0 { - for _, addr := range signers { - random, err := contracts.GetRandomizeFromContract(client, addr) - if err != nil { - log.Error("Fail to get random m2 from contract.", "error", err) - } - candidates = append(candidates, random) - } - - // Get randomize m2 list. - m2, err := contracts.GenM2FromRandomize(candidates, lenSigners) - if err != nil { - log.Error("Can not get m2 from randomize SC", "error", err) - } - if len(m2) > 0 { - header.Validators = contracts.BuildValidatorFromM2(m2) - log.Debug("New set Validators", "m2", m2, "number", header.Number.Uint64()) - } + validators, err := GetValidators(eth.blockchain, signers) + if err != nil { + return err } + header.Validators = validators } return nil } @@ -330,6 +308,19 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return nil } + c.VerifyValidators = func(header *types.Header, signers []common.Address) error { + number := header.Number.Int64() + if number > 0 && number%common.EpocBlockRandomize == 0 { + validators, err := GetValidators(eth.blockchain, signers) + if err != nil { + return err + } + if !bytes.Equal(header.Validators, validators) { + return posv.ErrInvalidCheckpointValidators + } + } + return nil + } } return eth, nil @@ -606,3 +597,37 @@ func (s *Ethereum) Stop() error { return nil } + +func GetValidators(bc *core.BlockChain, masternodes []common.Address) ([]byte, error) { + if bc.Config().Posv == nil { + return nil, core.ErrNotPoSV + } + client, err := bc.GetClient() + if err != nil { + return nil, err + } + // Check m2 exists on chaindb. + // Get secrets and opening at epoc block checkpoint. + + var candidates []int64 + if err != nil { + return nil, err + } + lenSigners := int64(len(masternodes)) + if lenSigners > 0 { + for _, addr := range masternodes { + random, err := contracts.GetRandomizeFromContract(client, addr) + if err != nil { + return nil, err + } + candidates = append(candidates, random) + } + // Get randomize m2 list. + m2, err := contracts.GenM2FromRandomize(candidates, lenSigners) + if err != nil { + return nil, err + } + return contracts.BuildValidatorFromM2(m2), nil + } + return nil, core.ErrNotFoundM1 +} From 6b43905a23b8b5e9a7823daeddfd42d0968cf8f7 Mon Sep 17 00:00:00 2001 From: Tuna Date: Tue, 23 Oct 2018 11:08:51 +0700 Subject: [PATCH 10/10] minor makeup --- consensus/posv/posv.go | 16 ++++++++-------- eth/backend.go | 15 +++++++++------ 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/consensus/posv/posv.go b/consensus/posv/posv.go index 303568fcc6..3f63137c36 100644 --- a/consensus/posv/posv.go +++ b/consensus/posv/posv.go @@ -215,10 +215,10 @@ type Posv struct { signFn clique.SignerFn // Signer function to authorize hashes with lock sync.RWMutex // Protects the signer fields - HookReward func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error - HookPenalty func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) - HookPrepare func(header *types.Header, signers []common.Address) error - VerifyValidators func(header *types.Header, signers []common.Address) error + HookReward func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error + HookPenalty func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) + HookValidator func(header *types.Header, signers []common.Address) error + HookVerifyMNs func(header *types.Header, signers []common.Address) error } // New creates a Posv proof-of-stake-voting consensus engine with the initial @@ -394,8 +394,8 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types. if !bytes.Equal(header.Extra[extraVanity:extraSuffix], byteMasterNodes) { return errInvalidCheckpointSigners } - if c.VerifyValidators != nil { - err := c.VerifyValidators(header, signers) + if c.HookVerifyMNs != nil { + err := c.HookVerifyMNs(header, signers) if err != nil { return err } @@ -705,8 +705,8 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error if header.Time.Int64() < time.Now().Unix() { header.Time = big.NewInt(time.Now().Unix()) } - if c.HookPrepare != nil { - c.HookPrepare(header, signers) + if c.HookValidator != nil { + c.HookValidator(header, signers) if err != nil { return err } diff --git a/eth/backend.go b/eth/backend.go index f0ed406c0f..be7e7275df 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -190,7 +190,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { if eth.chainConfig.Posv != nil { c := eth.engine.(*posv.Posv) - // Inject hook for send tx sign to smartcontract after insert block into chain. + // Hook sends tx sign to smartcontract after inserting block to chain. importedHook := func(block *types.Block) { snap, err := c.GetSnapshot(eth.blockchain, block.Header()) if err != nil { @@ -210,8 +210,8 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } eth.protocolManager.fetcher.SetImportedHook(importedHook) - // Hook will process when preparing block. - c.HookPrepare = func(header *types.Header, signers []common.Address) error { + // Hook prepares validators M2 for the current epoch + c.HookValidator = func(header *types.Header, signers []common.Address) error { number := header.Number.Int64() if number > 0 && number%common.EpocBlockRandomize == 0 { validators, err := GetValidators(eth.blockchain, signers) @@ -222,7 +222,8 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } return nil } - // Hook penalty. + + // Hook scans for bad masternodes and decide to penalty them c.HookPenalty = func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) { client, err := eth.blockchain.GetClient() if err != nil { @@ -263,7 +264,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return []common.Address{}, nil } - // Hook reward for posv validator. + // Hook calculates reward for masternodes c.HookReward = func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error { client, err := eth.blockchain.GetClient() if err != nil { @@ -308,7 +309,9 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return nil } - c.VerifyValidators = func(header *types.Header, signers []common.Address) error { + + // Hook verifies masternodes set + c.HookVerifyMNs = func(header *types.Header, signers []common.Address) error { number := header.Number.Int64() if number > 0 && number%common.EpocBlockRandomize == 0 { validators, err := GetValidators(eth.blockchain, signers)