From 780fc83a6fd60ed5cdda66229cf821dc26dc84ec Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 6 Sep 2017 14:39:21 +0200 Subject: [PATCH] les: implement GetTxStatusMsg and SendTxV2Msg --- core/database_util.go | 10 ++-- core/tx_list.go | 10 ++-- core/tx_pool.go | 106 +++++++++++++++++++++++++++++++++--------- les/handler.go | 96 +++++++++++++++++++++++++++++++++++--- les/peer.go | 15 +++++- les/protocol.go | 5 +- 6 files changed, 203 insertions(+), 39 deletions(-) diff --git a/core/database_util.go b/core/database_util.go index fb3d1a1e8a..c6b125dae9 100644 --- a/core/database_util.go +++ b/core/database_util.go @@ -74,9 +74,9 @@ var ( preimageHitCounter = metrics.NewCounter("db/preimage/hits") ) -// txLookupEntry is a positional metadata to help looking up the data content of +// TxLookupEntry is a positional metadata to help looking up the data content of // a transaction or receipt given only its hash. -type txLookupEntry struct { +type TxLookupEntry struct { BlockHash common.Hash BlockIndex uint64 Index uint64 @@ -260,7 +260,7 @@ func GetTxLookupEntry(db DatabaseReader, hash common.Hash) (common.Hash, uint64, return common.Hash{}, 0, 0 } // Parse and return the contents of the lookup entry - var entry txLookupEntry + var entry TxLookupEntry if err := rlp.DecodeBytes(data, &entry); err != nil { log.Error("Invalid lookup entry RLP", "hash", hash, "err", err) return common.Hash{}, 0, 0 @@ -296,7 +296,7 @@ func GetTransaction(db DatabaseReader, hash common.Hash) (*types.Transaction, co if len(data) == 0 { return nil, common.Hash{}, 0, 0 } - var entry txLookupEntry + var entry TxLookupEntry if err := rlp.DecodeBytes(data, &entry); err != nil { return nil, common.Hash{}, 0, 0 } @@ -464,7 +464,7 @@ func WriteBlockReceipts(db ethdb.Putter, hash common.Hash, number uint64, receip func WriteTxLookupEntries(db ethdb.Putter, block *types.Block) error { // Iterate over each transaction and encode its metadata for i, tx := range block.Transactions() { - entry := txLookupEntry{ + entry := TxLookupEntry{ BlockHash: block.Hash(), BlockIndex: block.NumberU64(), Index: uint64(i), diff --git a/core/tx_list.go b/core/tx_list.go index 2935929d72..94721aa5f3 100644 --- a/core/tx_list.go +++ b/core/tx_list.go @@ -384,13 +384,13 @@ func (h *priceHeap) Pop() interface{} { // txPricedList is a price-sorted heap to allow operating on transactions pool // contents in a price-incrementing way. type txPricedList struct { - all *map[common.Hash]*types.Transaction // Pointer to the map of all transactions - items *priceHeap // Heap of prices of all the stored transactions - stales int // Number of stale price points to (re-heap trigger) + all *map[common.Hash]txLookupRec // Pointer to the map of all transactions + items *priceHeap // Heap of prices of all the stored transactions + stales int // Number of stale price points to (re-heap trigger) } // newTxPricedList creates a new price-sorted transaction heap. -func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList { +func newTxPricedList(all *map[common.Hash]txLookupRec) *txPricedList { return &txPricedList{ all: all, items: new(priceHeap), @@ -416,7 +416,7 @@ func (l *txPricedList) Removed() { l.stales, l.items = 0, &reheap for _, tx := range *l.all { - *l.items = append(*l.items, tx) + *l.items = append(*l.items, tx.tx) } heap.Init(l.items) } diff --git a/core/tx_pool.go b/core/tx_pool.go index 0ad7651795..0130eb5713 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -192,17 +192,22 @@ type TxPool struct { locals *accountSet // Set of local transaction to exepmt from evicion rules journal *txJournal // Journal of local transaction to back up to disk - pending map[common.Address]*txList // All currently processable transactions - queue map[common.Address]*txList // Queued but non-processable transactions - beats map[common.Address]time.Time // Last heartbeat from each known account - all map[common.Hash]*types.Transaction // All transactions to allow lookups - priced *txPricedList // All transactions sorted by price + pending map[common.Address]*txList // All currently processable transactions + queue map[common.Address]*txList // Queued but non-processable transactions + beats map[common.Address]time.Time // Last heartbeat from each known account + all map[common.Hash]txLookupRec // All transactions to allow lookups + priced *txPricedList // All transactions sorted by price wg sync.WaitGroup // for shutdown sync homestead bool } +type txLookupRec struct { + tx *types.Transaction + pending bool +} + // NewTxPool creates a new transaction pool to gather, sort and filter inbound // trnsactions from the network. func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain blockChain) *TxPool { @@ -218,7 +223,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block pending: make(map[common.Address]*txList), queue: make(map[common.Address]*txList), beats: make(map[common.Address]time.Time), - all: make(map[common.Hash]*types.Transaction), + all: make(map[common.Hash]txLookupRec), chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), gasPrice: new(big.Int).SetUint64(config.PriceLimit), } @@ -594,7 +599,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { // If the transaction is already known, discard it hash := tx.Hash() - if pool.all[hash] != nil { + if _, ok := pool.all[hash]; ok { log.Trace("Discarding already known transaction", "hash", hash) return false, fmt.Errorf("known transaction: %x", hash) } @@ -635,7 +640,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { pool.priced.Removed() pendingReplaceCounter.Inc(1) } - pool.all[tx.Hash()] = tx + pool.all[tx.Hash()] = txLookupRec{tx, false} pool.priced.Put(tx) pool.journalTx(from, tx) @@ -678,7 +683,7 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er pool.priced.Removed() queuedReplaceCounter.Inc(1) } - pool.all[hash] = tx + pool.all[hash] = txLookupRec{tx, false} pool.priced.Put(tx) return old != nil, nil } @@ -721,10 +726,13 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T pendingReplaceCounter.Inc(1) } - // Failsafe to work around direct pending inserts (tests) - if pool.all[hash] == nil { - pool.all[hash] = tx + if pool.all[hash].tx == nil { + // Failsafe to work around direct pending inserts (tests) + pool.all[hash] = txLookupRec{tx, true} pool.priced.Put(tx) + } else { + // set pending flag to true + pool.all[hash] = txLookupRec{tx, true} } // Set the potentially new pending nonce and notify any subsystems of the new tx pool.beats[addr] = time.Now() @@ -750,14 +758,16 @@ func (pool *TxPool) AddRemote(tx *types.Transaction) error { // marking the senders as a local ones in the mean time, ensuring they go around // the local pricing constraints. func (pool *TxPool) AddLocals(txs []*types.Transaction) error { - return pool.addTxs(txs, !pool.config.NoLocals) + pool.addTxs(txs, !pool.config.NoLocals) + return nil } // AddRemotes enqueues a batch of transactions into the pool if they are valid. // If the senders are not among the locally tracked ones, full pricing constraints // will apply. func (pool *TxPool) AddRemotes(txs []*types.Transaction) error { - return pool.addTxs(txs, false) + pool.addTxs(txs, false) + return nil } // addTx enqueues a single transaction into the pool if it is valid. @@ -779,7 +789,7 @@ func (pool *TxPool) addTx(tx *types.Transaction, local bool) error { } // addTxs attempts to queue a batch of transactions if they are valid. -func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error { +func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) []error { pool.mu.Lock() defer pool.mu.Unlock() @@ -788,11 +798,13 @@ func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error { // addTxsLocked attempts to queue a batch of transactions if they are valid, // whilst assuming the transaction pool lock is already held. -func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) error { +func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) []error { // Add the batch of transaction, tracking the accepted ones dirty := make(map[common.Address]struct{}) - for _, tx := range txs { - if replace, err := pool.add(tx, local); err == nil { + txErr := make([]error, len(txs)) + for i, tx := range txs { + var replace bool + if replace, txErr[i] = pool.add(tx, local); txErr[i] == nil { if !replace { from, _ := types.Sender(pool.signer, tx) // already validated dirty[from] = struct{}{} @@ -807,7 +819,58 @@ func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) error { } pool.promoteExecutables(addrs) } - return nil + return txErr +} + +// TxStatusData is returned by AddOrGetTxStatus for each transaction +type TxStatusData struct { + Status uint + Data []byte +} + +const ( + TxStatusUnknown = iota + TxStatusQueued + TxStatusPending + TxStatusIncluded // Data contains a TxChainPos struct + TxStatusError // Data contains the error string +) + +// AddOrGetTxStatus returns the status (unknown/pending/queued) of a batch of transactions +// identified by their hashes in txHashes. Optionally the transactions themselves can be +// passed too in txs, in which case the function will try adding the previously unknown ones +// to the pool. If a new transaction cannot be added, TxStatusError is returned. Adding already +// known transactions will return their previous status. +// If txs is specified, txHashes is still required and has to match the transactions in txs. + +// Note: TxStatusIncluded is never returned by this function since the pool does not track +// mined transactions. Included status can be checked by the caller (as it happens in the +// LES protocol manager) +func (pool *TxPool) AddOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []TxStatusData { + status := make([]TxStatusData, len(txHashes)) + if txs != nil { + if len(txs) != len(txHashes) { + panic(nil) + } + txErr := pool.addTxs(txs, false) + for i, err := range txErr { + if err != nil { + status[i] = TxStatusData{TxStatusError, ([]byte)(err.Error())} + } + } + } + + for i, hash := range txHashes { + r, ok := pool.all[hash] + if ok { + if r.pending { + status[i] = TxStatusData{TxStatusPending, nil} + } else { + status[i] = TxStatusData{TxStatusQueued, nil} + } + } + } + return status } // Get returns a transaction if it is contained in the pool @@ -816,17 +879,18 @@ func (pool *TxPool) Get(hash common.Hash) *types.Transaction { pool.mu.RLock() defer pool.mu.RUnlock() - return pool.all[hash] + return pool.all[hash].tx } // removeTx removes a single transaction from the queue, moving all subsequent // transactions back to the future queue. func (pool *TxPool) removeTx(hash common.Hash) { // Fetch the transaction we wish to delete - tx, ok := pool.all[hash] + txl, ok := pool.all[hash] if !ok { return } + tx := txl.tx addr, _ := types.Sender(pool.signer, tx) // already validated during insertion // Remove it from the list of known transactions diff --git a/les/handler.go b/les/handler.go index bff502e1a8..287eca99b8 100644 --- a/les/handler.go +++ b/les/handler.go @@ -59,6 +59,7 @@ const ( MaxProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request MaxPPTProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request MaxTxSend = 64 // Amount of transactions to be send per request + MaxTxStatus = 256 // Amount of transactions to queried per request disableClientRemovePeer = false ) @@ -88,8 +89,7 @@ type BlockChain interface { } type txPool interface { - // AddRemotes should add the given transactions to the pool. - AddRemotes([]*types.Transaction) error + AddOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []core.TxStatusData } type ProtocolManager struct { @@ -318,7 +318,7 @@ func (pm *ProtocolManager) handle(p *peer) error { } } -var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsV1Msg, SendTxMsg, GetHeaderProofsMsg, GetProofsV2Msg, GetPPTProofsMsg} +var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsV1Msg, SendTxMsg, SendTxV2Msg, GetTxStatusMsg, GetHeaderProofsMsg, GetProofsV2Msg, GetPPTProofsMsg} // handleMsg is invoked whenever an inbound message is received from a remote // peer. The remote connection is torn down upon returning any error. @@ -972,7 +972,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { case SendTxMsg: if pm.txpool == nil { - return errResp(ErrUnexpectedResponse, "") + return errResp(ErrRequestRejected, "") } // Transactions arrived, parse all of them and deliver to the pool var txs []*types.Transaction @@ -984,13 +984,82 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrRequestRejected, "") } - if err := pm.txpool.AddRemotes(txs); err != nil { - return errResp(ErrUnexpectedResponse, "msg: %v", err) + txHashes := make([]common.Hash, len(txs)) + for i, tx := range txs { + txHashes[i] = tx.Hash() } + pm.addOrGetTxStatus(txs, txHashes) _, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) + case SendTxV2Msg: + if pm.txpool == nil { + return errResp(ErrRequestRejected, "") + } + // Transactions arrived, parse all of them and deliver to the pool + var req struct { + ReqID uint64 + Txs []*types.Transaction + } + if err := msg.Decode(&req); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + reqCnt := len(req.Txs) + if reject(uint64(reqCnt), MaxTxSend) { + return errResp(ErrRequestRejected, "") + } + + txHashes := make([]common.Hash, len(req.Txs)) + for i, tx := range req.Txs { + txHashes[i] = tx.Hash() + } + + res := pm.addOrGetTxStatus(req.Txs, txHashes) + + bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) + return p.SendTxStatus(req.ReqID, bv, res) + + case GetTxStatusMsg: + if pm.txpool == nil { + return errResp(ErrUnexpectedResponse, "") + } + // Transactions arrived, parse all of them and deliver to the pool + var req struct { + ReqID uint64 + TxHashes []common.Hash + } + if err := msg.Decode(&req); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + reqCnt := len(req.TxHashes) + if reject(uint64(reqCnt), MaxTxStatus) { + return errResp(ErrRequestRejected, "") + } + + res := pm.addOrGetTxStatus(nil, req.TxHashes) + + bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) + return p.SendTxStatus(req.ReqID, bv, res) + + case TxStatusMsg: + if pm.odr == nil { + return errResp(ErrUnexpectedResponse, "") + } + + p.Log().Trace("Received tx status response") + var resp struct { + ReqID, BV uint64 + Status []core.TxStatusData + } + if err := msg.Decode(&resp); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + + p.fcServer.GotReply(resp.ReqID, resp.BV) + default: p.Log().Trace("Received unknown message", "code", msg.Code) return errResp(ErrInvalidMsgCode, "%v", msg.Code) @@ -1034,6 +1103,21 @@ func (pm *ProtocolManager) getPPTAuxData(req PPTReq) []byte { return nil } +func (pm *ProtocolManager) addOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []core.TxStatusData { + status := pm.txpool.AddOrGetTxStatus(txs, txHashes) + for i, _ := range status { + blockHash, blockNum, txIndex := core.GetTxLookupEntry(pm.chainDb, txHashes[i]) + if blockHash != (common.Hash{}) { + enc, err := rlp.EncodeToBytes(core.TxLookupEntry{BlockHash: blockHash, BlockIndex: blockNum, Index: txIndex}) + if err != nil { + panic(err) + } + status[i] = core.TxStatusData{Status: core.TxStatusIncluded, Data: enc} + } + } + return status +} + // NodeInfo retrieves some protocol metadata about the running host node. func (self *ProtocolManager) NodeInfo() *eth.EthNodeInfo { return ð.EthNodeInfo{ diff --git a/les/peer.go b/les/peer.go index 1c31456b35..aa4120f836 100644 --- a/les/peer.go +++ b/les/peer.go @@ -27,6 +27,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" @@ -231,6 +232,11 @@ func (p *peer) SendPPTProofs(reqID, bv uint64, resp PPTResps) error { return sendResponse(p.rw, PPTProofsMsg, reqID, bv, resp) } +// SendTxStatus sends a batch of transaction status records, corresponding to the ones requested. +func (p *peer) SendTxStatus(reqID, bv uint64, status []core.TxStatusData) error { + return sendResponse(p.rw, TxStatusMsg, reqID, bv, status) +} + // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the // specified header query, based on the hash of an origin block. func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amount int, skip int, reverse bool) error { @@ -303,7 +309,14 @@ func (p *peer) RequestPPTProofs(reqID, cost uint64, reqs []PPTReq) error { func (p *peer) SendTxs(reqID, cost uint64, txs types.Transactions) error { p.Log().Debug("Fetching batch of transactions", "count", len(txs)) - return p2p.Send(p.rw, SendTxMsg, txs) + switch p.version { + case lpv1: + return p2p.Send(p.rw, SendTxMsg, txs) // old message format does not include reqID + case lpv2: + return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs) + default: + panic(nil) + } } type keyValueEntry struct { diff --git a/les/protocol.go b/les/protocol.go index a00d624e1b..19e0912982 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -45,7 +45,7 @@ var ( ) // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = map[uint]uint64{lpv1: 15, lpv2: 19} +var ProtocolLengths = map[uint]uint64{lpv1: 15, lpv2: 22} const ( NetworkId = 1 @@ -75,6 +75,9 @@ const ( ProofsV2Msg = 0x10 GetPPTProofsMsg = 0x11 PPTProofsMsg = 0x12 + SendTxV2Msg = 0x13 + GetTxStatusMsg = 0x14 + TxStatusMsg = 0x15 ) type errCode int