mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-06-19 21:31:37 +00:00
all: change TxPreEvent to TxsPreEvent (#16720)
This commit is contained in:
Daniel Liu
parent
63cc3336c3
commit
1c9f1d8279
19 changed files with 171 additions and 123 deletions
|
|
@ -553,7 +553,7 @@ func (fb *filterBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*ty
|
|||
return logs, nil
|
||||
}
|
||||
|
||||
func (fb *filterBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
func (fb *filterBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return event.NewSubscription(func(quit <-chan struct{}) error {
|
||||
<-quit
|
||||
return nil
|
||||
|
|
|
|||
|
|
@ -21,8 +21,8 @@ import (
|
|||
"github.com/XinFinOrg/XDPoSChain/core/types"
|
||||
)
|
||||
|
||||
// TxPreEvent is posted when a transaction enters the transaction pool.
|
||||
type TxPreEvent struct{ Tx *types.Transaction }
|
||||
// NewTxsEvent is posted when a batch of transactions enter the transaction pool.
|
||||
type NewTxsEvent struct{ Txs []*types.Transaction }
|
||||
|
||||
// OrderTxPreEvent is posted when a order transaction enters the order transaction pool.
|
||||
type OrderTxPreEvent struct{ Tx *types.OrderTransaction }
|
||||
|
|
@ -41,9 +41,6 @@ type PendingStateEvent struct{}
|
|||
// NewMinedBlockEvent is posted when a block has been imported.
|
||||
type NewMinedBlockEvent struct{ Block *types.Block }
|
||||
|
||||
// RemovedTransactionEvent is posted when a reorg happens
|
||||
type RemovedTransactionEvent struct{ Txs types.Transactions }
|
||||
|
||||
// RemovedLogsEvent is posted when a reorg happens
|
||||
type RemovedLogsEvent struct{ Logs []*types.Log }
|
||||
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ func newTxJournal(path string) *txJournal {
|
|||
|
||||
// load parses a transaction journal dump from disk, loading its contents into
|
||||
// the specified pool.
|
||||
func (journal *txJournal) load(add func(*types.Transaction) error) error {
|
||||
func (journal *txJournal) load(add func([]*types.Transaction) []error) error {
|
||||
// Skip the parsing if the journal file doens't exist at all
|
||||
if _, err := os.Stat(journal.path); os.IsNotExist(err) {
|
||||
return nil
|
||||
|
|
@ -76,7 +76,21 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error {
|
|||
stream := rlp.NewStream(input, 0)
|
||||
total, dropped := 0, 0
|
||||
|
||||
var failure error
|
||||
// Create a method to load a limited batch of transactions and bump the
|
||||
// appropriate progress counters. Then use this method to load all the
|
||||
// journalled transactions in small-ish batches.
|
||||
loadBatch := func(txs types.Transactions) {
|
||||
for _, err := range add(txs) {
|
||||
if err != nil {
|
||||
log.Debug("Failed to add journaled transaction", "err", err)
|
||||
dropped++
|
||||
}
|
||||
}
|
||||
}
|
||||
var (
|
||||
failure error
|
||||
batch types.Transactions
|
||||
)
|
||||
for {
|
||||
// Parse the next transaction and terminate on error
|
||||
tx := new(types.Transaction)
|
||||
|
|
@ -84,14 +98,16 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error {
|
|||
if err != io.EOF {
|
||||
failure = err
|
||||
}
|
||||
if batch.Len() > 0 {
|
||||
loadBatch(batch)
|
||||
}
|
||||
break
|
||||
}
|
||||
// Import the transaction and bump the appropriate progress counters
|
||||
// New transaction parsed, queue up for later, import if threnshold is reached
|
||||
total++
|
||||
if err = add(tx); err != nil {
|
||||
log.Debug("Failed to add journaled transaction", "err", err)
|
||||
dropped++
|
||||
continue
|
||||
if batch = append(batch, tx); batch.Len() > 1024 {
|
||||
loadBatch(batch)
|
||||
batch = batch[:0]
|
||||
}
|
||||
}
|
||||
log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)
|
||||
|
|
|
|||
|
|
@ -40,8 +40,6 @@ import (
|
|||
const (
|
||||
// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
|
||||
chainHeadChanSize = 10
|
||||
// rmTxChanSize is the size of channel listening to RemovedTransactionEvent.
|
||||
rmTxChanSize = 10
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
@ -264,7 +262,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
|
|||
if !config.NoLocals && config.Journal != "" {
|
||||
pool.journal = newTxJournal(config.Journal)
|
||||
|
||||
if err := pool.journal.load(pool.AddLocal); err != nil {
|
||||
if err := pool.journal.load(pool.AddLocals); err != nil {
|
||||
log.Warn("Failed to load transaction journal", "err", err)
|
||||
}
|
||||
if err := pool.journal.rotate(pool.local()); err != nil {
|
||||
|
|
@ -471,9 +469,9 @@ func (pool *TxPool) Stop() {
|
|||
log.Info("Transaction pool stopped")
|
||||
}
|
||||
|
||||
// SubscribeTxPreEvent registers a subscription of TxPreEvent and
|
||||
// SubscribeNewTxsEvent registers a subscription of NewTxsEvent and
|
||||
// starts sending event to the given channel.
|
||||
func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription {
|
||||
func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- NewTxsEvent) event.Subscription {
|
||||
return pool.scope.Track(pool.txFeed.Subscribe(ch))
|
||||
}
|
||||
|
||||
|
|
@ -758,7 +756,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
|
|||
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
|
||||
|
||||
// We've directly injected a replacement transaction, notify subsystems
|
||||
go pool.txFeed.Send(TxPreEvent{tx})
|
||||
go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}})
|
||||
|
||||
return old != nil, nil
|
||||
}
|
||||
|
|
@ -815,10 +813,11 @@ func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) {
|
|||
}
|
||||
}
|
||||
|
||||
// promoteTx adds a transaction to the pending (processable) list of transactions.
|
||||
// promoteTx adds a transaction to the pending (processable) list of transactions
|
||||
// and returns whether it was inserted or an older was better.
|
||||
//
|
||||
// Note, this method assumes the pool lock is held!
|
||||
func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) {
|
||||
func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) bool {
|
||||
// Try to insert the transaction into the pending queue
|
||||
if pool.pending[addr] == nil {
|
||||
pool.pending[addr] = newTxList(true)
|
||||
|
|
@ -832,7 +831,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
|
|||
pool.priced.Removed()
|
||||
|
||||
pendingDiscardCounter.Inc(1)
|
||||
return
|
||||
return false
|
||||
}
|
||||
// Otherwise discard any previous transaction and mark this
|
||||
if old != nil {
|
||||
|
|
@ -850,7 +849,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
|
|||
pool.beats[addr] = time.Now()
|
||||
pool.pendingState.SetNonce(addr, tx.Nonce()+1)
|
||||
|
||||
go pool.txFeed.Send(TxPreEvent{tx})
|
||||
return true
|
||||
}
|
||||
|
||||
func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction) (bool, error) {
|
||||
|
|
@ -883,7 +882,7 @@ func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction)
|
|||
// 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)
|
||||
go pool.txFeed.Send(TxPreEvent{tx})
|
||||
go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
|
|
@ -1050,6 +1049,9 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
|
|||
log.Debug("end promoteExecutables", "time", common.PrettyDuration(time.Since(start)))
|
||||
}(time.Now())
|
||||
|
||||
// Track the promoted transactions to broadcast them at once
|
||||
var promoted []*types.Transaction
|
||||
|
||||
// Gather all the accounts potentially needing updates
|
||||
if accounts == nil {
|
||||
accounts = make([]common.Address, 0, len(pool.queue))
|
||||
|
|
@ -1086,8 +1088,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
|
|||
// Gather all executable transactions and promote them
|
||||
for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) {
|
||||
hash := tx.Hash()
|
||||
log.Trace("Promoting queued transaction", "hash", hash)
|
||||
pool.promoteTx(addr, hash, tx)
|
||||
if pool.promoteTx(addr, hash, tx) {
|
||||
log.Trace("Promoting queued transaction", "hash", hash)
|
||||
promoted = append(promoted, tx)
|
||||
}
|
||||
}
|
||||
// Drop all transactions over the allowed limit
|
||||
if !pool.locals.contains(addr) {
|
||||
|
|
@ -1104,6 +1108,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
|
|||
delete(pool.queue, addr)
|
||||
}
|
||||
}
|
||||
// Notify subsystem for new promoted transactions.
|
||||
if len(promoted) > 0 {
|
||||
pool.txFeed.Send(NewTxsEvent{promoted})
|
||||
}
|
||||
// If the pending limit is overflown, start equalizing allowances
|
||||
pending := uint64(0)
|
||||
for _, list := range pool.pending {
|
||||
|
|
|
|||
|
|
@ -136,21 +136,27 @@ func validateTxPoolInternals(pool *TxPool) error {
|
|||
|
||||
// validateEvents checks that the correct number of transaction addition events
|
||||
// were fired on the pool's event feed.
|
||||
func validateEvents(events chan TxPreEvent, count int) error {
|
||||
for i := 0; i < count; i++ {
|
||||
func validateEvents(events chan NewTxsEvent, count int) error {
|
||||
var received []*types.Transaction
|
||||
|
||||
for len(received) < count {
|
||||
select {
|
||||
case <-events:
|
||||
case ev := <-events:
|
||||
received = append(received, ev.Txs...)
|
||||
case <-time.After(time.Second):
|
||||
return fmt.Errorf("event #%d not fired", i)
|
||||
return fmt.Errorf("event #%d not fired", len(received))
|
||||
}
|
||||
}
|
||||
if len(received) > count {
|
||||
return fmt.Errorf("more than %d events fired: %v", count, received[count:])
|
||||
}
|
||||
select {
|
||||
case tx := <-events:
|
||||
return fmt.Errorf("more than %d events fired: %v", count, tx.Tx)
|
||||
case ev := <-events:
|
||||
return fmt.Errorf("more than %d events fired: %v", count, ev.Txs)
|
||||
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
// This branch should be "default", but it's a data race between goroutines,
|
||||
// reading the event channel and pushng into it, so better wait a bit ensuring
|
||||
// reading the event channel and pushing into it, so better wait a bit ensuring
|
||||
// really nothing gets injected.
|
||||
}
|
||||
return nil
|
||||
|
|
@ -697,7 +703,7 @@ func TestTransactionGapFilling(t *testing.T) {
|
|||
pool.currentState.AddBalance(account, big.NewInt(1000000))
|
||||
|
||||
// Keep track of transaction events to ensure all executables get announced
|
||||
events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue+5)
|
||||
events := make(chan NewTxsEvent, testTxPoolConfig.AccountQueue+5)
|
||||
sub := pool.txFeed.Subscribe(events)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
|
|
@ -954,7 +960,7 @@ func TestTransactionPendingLimiting(t *testing.T) {
|
|||
pool.currentState.AddBalance(account, big.NewInt(1000000))
|
||||
testTxPoolConfig.AccountQueue = 10
|
||||
// Keep track of transaction events to ensure all executables get announced
|
||||
events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue)
|
||||
events := make(chan NewTxsEvent, testTxPoolConfig.AccountQueue)
|
||||
sub := pool.txFeed.Subscribe(events)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
|
|
@ -1181,7 +1187,7 @@ func TestTransactionPoolRepricing(t *testing.T) {
|
|||
defer pool.Stop()
|
||||
|
||||
// Keep track of transaction events to ensure all executables get announced
|
||||
events := make(chan TxPreEvent, 32)
|
||||
events := make(chan NewTxsEvent, 32)
|
||||
sub := pool.txFeed.Subscribe(events)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
|
|
@ -1370,7 +1376,7 @@ func TestTransactionPoolUnderpricing(t *testing.T) {
|
|||
defer pool.Stop()
|
||||
|
||||
// Keep track of transaction events to ensure all executables get announced
|
||||
events := make(chan TxPreEvent, 32)
|
||||
events := make(chan NewTxsEvent, 32)
|
||||
sub := pool.txFeed.Subscribe(events)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
|
|
@ -1468,7 +1474,7 @@ func TestTransactionReplacement(t *testing.T) {
|
|||
defer pool.Stop()
|
||||
|
||||
// Keep track of transaction events to ensure all executables get announced
|
||||
events := make(chan TxPreEvent, 32)
|
||||
events := make(chan NewTxsEvent, 32)
|
||||
sub := pool.txFeed.Subscribe(events)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
|
|
|
|||
|
|
@ -322,8 +322,8 @@ func (b *EthApiBackend) OrderStats() (pending int, queued int) {
|
|||
return b.eth.txPool.Stats()
|
||||
}
|
||||
|
||||
func (b *EthApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
return b.eth.TxPool().SubscribeTxPreEvent(ch)
|
||||
func (b *EthApiBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return b.eth.TxPool().SubscribeNewTxsEvent(ch)
|
||||
}
|
||||
|
||||
func (b *EthApiBackend) Downloader() *downloader.Downloader {
|
||||
|
|
|
|||
|
|
@ -112,8 +112,8 @@ func (api *PublicFilterAPI) timeoutLoop() {
|
|||
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter
|
||||
func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
|
||||
var (
|
||||
pendingTxs = make(chan common.Hash)
|
||||
pendingTxSub = api.events.SubscribePendingTxEvents(pendingTxs)
|
||||
pendingTxs = make(chan []common.Hash)
|
||||
pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
|
||||
)
|
||||
|
||||
api.filtersMu.Lock()
|
||||
|
|
@ -126,7 +126,7 @@ func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
|
|||
case ph := <-pendingTxs:
|
||||
api.filtersMu.Lock()
|
||||
if f, found := api.filters[pendingTxSub.ID]; found {
|
||||
f.hashes = append(f.hashes, ph)
|
||||
f.hashes = append(f.hashes, ph...)
|
||||
}
|
||||
api.filtersMu.Unlock()
|
||||
case <-pendingTxSub.Err():
|
||||
|
|
@ -152,13 +152,17 @@ func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Su
|
|||
rpcSub := notifier.CreateSubscription()
|
||||
|
||||
go func() {
|
||||
txHashes := make(chan common.Hash)
|
||||
pendingTxSub := api.events.SubscribePendingTxEvents(txHashes)
|
||||
txHashes := make(chan []common.Hash, 128)
|
||||
pendingTxSub := api.events.SubscribePendingTxs(txHashes)
|
||||
|
||||
for {
|
||||
select {
|
||||
case h := <-txHashes:
|
||||
notifier.Notify(rpcSub.ID, h)
|
||||
case hashes := <-txHashes:
|
||||
// To keep the original behaviour, send a single tx hash in one notification.
|
||||
// TODO(rjl493456442) Send a batch of tx hashes in one notification
|
||||
for _, h := range hashes {
|
||||
notifier.Notify(rpcSub.ID, h)
|
||||
}
|
||||
case <-rpcSub.Err():
|
||||
pendingTxSub.Unsubscribe()
|
||||
return
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ type Backend interface {
|
|||
GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
|
||||
GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
|
||||
|
||||
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
||||
SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
|
||||
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
|
||||
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
|
||||
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
|
||||
|
|
|
|||
|
|
@ -57,8 +57,7 @@ const (
|
|||
)
|
||||
|
||||
const (
|
||||
|
||||
// txChanSize is the size of channel listening to TxPreEvent.
|
||||
// txChanSize is the size of channel listening to NewTxsEvent.
|
||||
// The number is referenced from the size of tx pool.
|
||||
txChanSize = 4096
|
||||
// rmLogsChanSize is the size of channel listening to RemovedLogsEvent.
|
||||
|
|
@ -79,7 +78,7 @@ type subscription struct {
|
|||
created time.Time
|
||||
logsCrit ethereum.FilterQuery
|
||||
logs chan []*types.Log
|
||||
hashes chan common.Hash
|
||||
hashes chan []common.Hash
|
||||
headers chan *types.Header
|
||||
installed chan struct{} // closed when the filter is installed
|
||||
err chan error // closed when the filter is uninstalled
|
||||
|
|
@ -94,7 +93,7 @@ type EventSystem struct {
|
|||
lastHead *types.Header
|
||||
|
||||
// Subscriptions
|
||||
txSub event.Subscription // Subscription for new transaction event
|
||||
txsSub event.Subscription // Subscription for new transaction event
|
||||
logsSub event.Subscription // Subscription for new log event
|
||||
rmLogsSub event.Subscription // Subscription for removed log event
|
||||
chainSub event.Subscription // Subscription for new chain event
|
||||
|
|
@ -103,7 +102,7 @@ type EventSystem struct {
|
|||
// Channels
|
||||
install chan *subscription // install filter for event notification
|
||||
uninstall chan *subscription // remove filter for event notification
|
||||
txCh chan core.TxPreEvent // Channel to receive new transaction event
|
||||
txsCh chan core.NewTxsEvent // Channel to receive new transactions event
|
||||
logsCh chan []*types.Log // Channel to receive new log event
|
||||
rmLogsCh chan core.RemovedLogsEvent // Channel to receive removed log event
|
||||
chainCh chan core.ChainEvent // Channel to receive new chain event
|
||||
|
|
@ -122,14 +121,14 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
|
|||
lightMode: lightMode,
|
||||
install: make(chan *subscription),
|
||||
uninstall: make(chan *subscription),
|
||||
txCh: make(chan core.TxPreEvent, txChanSize),
|
||||
txsCh: make(chan core.NewTxsEvent, txChanSize),
|
||||
logsCh: make(chan []*types.Log, logsChanSize),
|
||||
rmLogsCh: make(chan core.RemovedLogsEvent, rmLogsChanSize),
|
||||
chainCh: make(chan core.ChainEvent, chainEvChanSize),
|
||||
}
|
||||
|
||||
// Subscribe events
|
||||
m.txSub = m.backend.SubscribeTxPreEvent(m.txCh)
|
||||
m.txsSub = m.backend.SubscribeNewTxsEvent(m.txsCh)
|
||||
m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
|
||||
m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
|
||||
m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
|
||||
|
|
@ -137,7 +136,7 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
|
|||
m.pendingLogSub = m.mux.Subscribe(core.PendingLogsEvent{})
|
||||
|
||||
// Make sure none of the subscriptions are empty
|
||||
if m.txSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil ||
|
||||
if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil ||
|
||||
m.pendingLogSub.Closed() {
|
||||
log.Crit("Subscribe for event system failed")
|
||||
}
|
||||
|
|
@ -240,7 +239,7 @@ func (es *EventSystem) subscribeMinedPendingLogs(crit ethereum.FilterQuery, logs
|
|||
logsCrit: crit,
|
||||
created: time.Now(),
|
||||
logs: logs,
|
||||
hashes: make(chan common.Hash),
|
||||
hashes: make(chan []common.Hash),
|
||||
headers: make(chan *types.Header),
|
||||
installed: make(chan struct{}),
|
||||
err: make(chan error),
|
||||
|
|
@ -257,7 +256,7 @@ func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*typ
|
|||
logsCrit: crit,
|
||||
created: time.Now(),
|
||||
logs: logs,
|
||||
hashes: make(chan common.Hash),
|
||||
hashes: make(chan []common.Hash),
|
||||
headers: make(chan *types.Header),
|
||||
installed: make(chan struct{}),
|
||||
err: make(chan error),
|
||||
|
|
@ -274,7 +273,7 @@ func (es *EventSystem) subscribePendingLogs(crit ethereum.FilterQuery, logs chan
|
|||
logsCrit: crit,
|
||||
created: time.Now(),
|
||||
logs: logs,
|
||||
hashes: make(chan common.Hash),
|
||||
hashes: make(chan []common.Hash),
|
||||
headers: make(chan *types.Header),
|
||||
installed: make(chan struct{}),
|
||||
err: make(chan error),
|
||||
|
|
@ -290,7 +289,7 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
|
|||
typ: BlocksSubscription,
|
||||
created: time.Now(),
|
||||
logs: make(chan []*types.Log),
|
||||
hashes: make(chan common.Hash),
|
||||
hashes: make(chan []common.Hash),
|
||||
headers: headers,
|
||||
installed: make(chan struct{}),
|
||||
err: make(chan error),
|
||||
|
|
@ -298,9 +297,9 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
|
|||
return es.subscribe(sub)
|
||||
}
|
||||
|
||||
// SubscribePendingTxEvents creates a subscription that writes transaction hashes for
|
||||
// SubscribePendingTxs creates a subscription that writes transaction hashes for
|
||||
// transactions that enter the transaction pool.
|
||||
func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscription {
|
||||
func (es *EventSystem) SubscribePendingTxs(hashes chan []common.Hash) *Subscription {
|
||||
sub := &subscription{
|
||||
id: rpc.NewID(),
|
||||
typ: PendingTransactionsSubscription,
|
||||
|
|
@ -347,9 +346,13 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
|
|||
}
|
||||
}
|
||||
}
|
||||
case core.TxPreEvent:
|
||||
case core.NewTxsEvent:
|
||||
hashes := make([]common.Hash, 0, len(e.Txs))
|
||||
for _, tx := range e.Txs {
|
||||
hashes = append(hashes, tx.Hash())
|
||||
}
|
||||
for _, f := range filters[PendingTransactionsSubscription] {
|
||||
f.hashes <- e.Tx.Hash()
|
||||
f.hashes <- hashes
|
||||
}
|
||||
case core.ChainEvent:
|
||||
for _, f := range filters[BlocksSubscription] {
|
||||
|
|
@ -445,7 +448,7 @@ func (es *EventSystem) eventLoop() {
|
|||
// Ensure all subscriptions get cleaned up
|
||||
defer func() {
|
||||
es.pendingLogSub.Unsubscribe()
|
||||
es.txSub.Unsubscribe()
|
||||
es.txsSub.Unsubscribe()
|
||||
es.logsSub.Unsubscribe()
|
||||
es.rmLogsSub.Unsubscribe()
|
||||
es.chainSub.Unsubscribe()
|
||||
|
|
@ -459,7 +462,7 @@ func (es *EventSystem) eventLoop() {
|
|||
for {
|
||||
select {
|
||||
// Handle subscribed events
|
||||
case ev := <-es.txCh:
|
||||
case ev := <-es.txsCh:
|
||||
es.broadcast(index, ev)
|
||||
case ev := <-es.logsCh:
|
||||
es.broadcast(index, ev)
|
||||
|
|
@ -494,7 +497,7 @@ func (es *EventSystem) eventLoop() {
|
|||
close(f.err)
|
||||
|
||||
// System stopped
|
||||
case <-es.txSub.Err():
|
||||
case <-es.txsSub.Err():
|
||||
return
|
||||
case <-es.logsSub.Err():
|
||||
return
|
||||
|
|
|
|||
|
|
@ -90,7 +90,7 @@ func (b *testBackend) GetLogs(ctx context.Context, blockHash common.Hash) ([][]*
|
|||
return logs, nil
|
||||
}
|
||||
|
||||
func (b *testBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
func (b *testBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return b.txFeed.Subscribe(ch)
|
||||
}
|
||||
|
||||
|
|
@ -226,10 +226,7 @@ func TestPendingTxFilter(t *testing.T) {
|
|||
fid0 := api.NewPendingTransactionFilter()
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
for _, tx := range transactions {
|
||||
ev := core.TxPreEvent{Tx: tx}
|
||||
txFeed.Send(ev)
|
||||
}
|
||||
txFeed.Send(core.NewTxsEvent{Txs: transactions})
|
||||
|
||||
timeout := time.Now().Add(1 * time.Second)
|
||||
for {
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ const (
|
|||
softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data.
|
||||
estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header
|
||||
|
||||
// txChanSize is the size of channel listening to TxPreEvent.
|
||||
// txChanSize is the size of channel listening to NewTxsEvent.
|
||||
// The number is referenced from the size of tx pool.
|
||||
txChanSize = 4096
|
||||
)
|
||||
|
|
@ -87,10 +87,10 @@ type ProtocolManager struct {
|
|||
SubProtocols []p2p.Protocol
|
||||
|
||||
eventMux *event.TypeMux
|
||||
txCh chan core.TxPreEvent
|
||||
txsCh chan core.NewTxsEvent
|
||||
orderTxCh chan core.OrderTxPreEvent
|
||||
lendingTxCh chan core.LendingTxPreEvent
|
||||
txSub event.Subscription
|
||||
txsSub event.Subscription
|
||||
orderTxSub event.Subscription
|
||||
lendingTxSub event.Subscription
|
||||
minedBlockSub *event.TypeMuxSubscription
|
||||
|
|
@ -292,8 +292,8 @@ func (pm *ProtocolManager) Start(maxPeers int) {
|
|||
pm.maxPeers = maxPeers
|
||||
|
||||
// broadcast transactions
|
||||
pm.txCh = make(chan core.TxPreEvent, txChanSize)
|
||||
pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh)
|
||||
pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
|
||||
pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)
|
||||
pm.orderTxCh = make(chan core.OrderTxPreEvent, txChanSize)
|
||||
if pm.orderpool != nil {
|
||||
pm.orderTxSub = pm.orderpool.SubscribeTxPreEvent(pm.orderTxCh)
|
||||
|
|
@ -317,7 +317,7 @@ func (pm *ProtocolManager) Start(maxPeers int) {
|
|||
func (pm *ProtocolManager) Stop() {
|
||||
log.Info("Stopping Ethereum protocol")
|
||||
|
||||
pm.txSub.Unsubscribe() // quits txBroadcastLoop
|
||||
pm.txsSub.Unsubscribe() // quits txBroadcastLoop
|
||||
if pm.orderTxSub != nil {
|
||||
pm.orderTxSub.Unsubscribe()
|
||||
}
|
||||
|
|
@ -941,16 +941,23 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
|
|||
}
|
||||
}
|
||||
|
||||
// BroadcastTx will propagate a transaction to all peers which are not known to
|
||||
// BroadcastTxs will propagate a batch of transactions to all peers which are not known to
|
||||
// already have the given transaction.
|
||||
func (pm *ProtocolManager) BroadcastTx(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.SendTransactions(types.Transactions{tx})
|
||||
func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) {
|
||||
var txset = make(map[*peer]types.Transactions)
|
||||
|
||||
// Broadcast transactions to a batch of peers not knowing about it
|
||||
for _, tx := range txs {
|
||||
peers := pm.peers.PeersWithoutTx(tx.Hash())
|
||||
for _, peer := range peers {
|
||||
txset[peer] = append(txset[peer], tx)
|
||||
}
|
||||
log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers))
|
||||
}
|
||||
// FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
|
||||
for peer, txs := range txset {
|
||||
peer.SendTransactions(txs)
|
||||
}
|
||||
log.Trace("Broadcast transaction", "hash", hash, "recipients", len(peers))
|
||||
}
|
||||
|
||||
// BroadcastVote will propagate a Vote to all peers which are not known to
|
||||
|
|
@ -1041,14 +1048,14 @@ func (self *ProtocolManager) minedBroadcastLoop() {
|
|||
}
|
||||
}
|
||||
|
||||
func (self *ProtocolManager) txBroadcastLoop() {
|
||||
func (pm *ProtocolManager) txBroadcastLoop() {
|
||||
for {
|
||||
select {
|
||||
case event := <-self.txCh:
|
||||
self.BroadcastTx(event.Tx.Hash(), event.Tx)
|
||||
case event := <-pm.txsCh:
|
||||
pm.BroadcastTxs(event.Txs)
|
||||
|
||||
// Err() channel will be closed when unsubscribing.
|
||||
case <-self.txSub.Err():
|
||||
// Err() channel will be closed when unsubscribing.
|
||||
case <-pm.txsSub.Err():
|
||||
return
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -126,7 +126,7 @@ func (p *testTxPool) Pending() (map[common.Address]types.Transactions, error) {
|
|||
return batches, nil
|
||||
}
|
||||
|
||||
func (p *testTxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
func (p *testTxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return p.txFeed.Subscribe(ch)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -110,9 +110,9 @@ type txPool interface {
|
|||
// The slice should be modifiable by the caller.
|
||||
Pending() (map[common.Address]types.Transactions, error)
|
||||
|
||||
// SubscribeTxPreEvent should return an event subscription of
|
||||
// TxPreEvent and send events to the given channel.
|
||||
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
||||
// SubscribeNewTxsEvent should return an event subscription of
|
||||
// NewTxsEvent and send events to the given channel.
|
||||
SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
|
||||
}
|
||||
|
||||
type orderPool interface {
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ func testRecvTransactions(t *testing.T, protocol int) {
|
|||
t.Errorf("added wrong tx hash: got %v, want %v", added[0].Hash(), tx.Hash())
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Errorf("no TxPreEvent received within 2 seconds")
|
||||
t.Errorf("no NewTxsEvent received within 2 seconds")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ const (
|
|||
// history request.
|
||||
historyUpdateRange = 50
|
||||
|
||||
// txChanSize is the size of channel listening to TxPreEvent.
|
||||
// txChanSize is the size of channel listening to NewTxsEvent.
|
||||
// The number is referenced from the size of tx pool.
|
||||
txChanSize = 4096
|
||||
// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
|
||||
|
|
@ -65,9 +65,9 @@ type consensusEngine interface {
|
|||
}
|
||||
|
||||
type txPool interface {
|
||||
// SubscribeTxPreEvent should return an event subscription of
|
||||
// TxPreEvent and send events to the given channel.
|
||||
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
||||
// SubscribeNewTxsEvent should return an event subscription of
|
||||
// NewTxsEvent and send events to the given channel.
|
||||
SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
|
||||
}
|
||||
|
||||
type blockChain interface {
|
||||
|
|
@ -208,8 +208,8 @@ func (s *Service) loop() {
|
|||
headSub := blockchain.SubscribeChainHeadEvent(chainHeadCh)
|
||||
defer headSub.Unsubscribe()
|
||||
|
||||
txEventCh := make(chan core.TxPreEvent, txChanSize)
|
||||
txSub := txpool.SubscribeTxPreEvent(txEventCh)
|
||||
txEventCh := make(chan core.NewTxsEvent, txChanSize)
|
||||
txSub := txpool.SubscribeNewTxsEvent(txEventCh)
|
||||
defer txSub.Unsubscribe()
|
||||
|
||||
// Forensics events
|
||||
|
|
|
|||
|
|
@ -79,7 +79,7 @@ type Backend interface {
|
|||
GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error)
|
||||
Stats() (pending int, queued int)
|
||||
TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions)
|
||||
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
||||
SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
|
||||
|
||||
// Order Pool Transaction
|
||||
SendOrderTx(ctx context.Context, signedTx *types.OrderTransaction) error
|
||||
|
|
|
|||
|
|
@ -218,8 +218,8 @@ func (b *LesApiBackend) OrderStats() (pending int, queued int) {
|
|||
return 0, 0
|
||||
}
|
||||
|
||||
func (b *LesApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
return b.eth.txPool.SubscribeTxPreEvent(ch)
|
||||
func (b *LesApiBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return b.eth.txPool.SubscribeNewTxsEvent(ch)
|
||||
}
|
||||
|
||||
func (b *LesApiBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
|
||||
|
|
|
|||
|
|
@ -74,10 +74,13 @@ type TxPool struct {
|
|||
//
|
||||
// Send instructs backend to forward new transactions
|
||||
// NewHead notifies backend about a new head after processed by the tx pool,
|
||||
// including mined and rolled back transactions since the last event
|
||||
//
|
||||
// including mined and rolled back transactions since the last event
|
||||
//
|
||||
// Discard notifies backend about transactions that should be discarded either
|
||||
// because they have been replaced by a re-send or because they have been mined
|
||||
// long ago and no rollback is expected
|
||||
//
|
||||
// because they have been replaced by a re-send or because they have been mined
|
||||
// long ago and no rollback is expected
|
||||
type TxRelayBackend interface {
|
||||
Send(txs types.Transactions)
|
||||
NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash)
|
||||
|
|
@ -321,9 +324,9 @@ func (pool *TxPool) Stop() {
|
|||
log.Info("Transaction pool stopped")
|
||||
}
|
||||
|
||||
// SubscribeTxPreEvent registers a subscription of core.TxPreEvent and
|
||||
// SubscribeNewTxsEvent registers a subscription of core.NewTxsEvent and
|
||||
// starts sending event to the given channel.
|
||||
func (pool *TxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
|
||||
func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
|
||||
return pool.scope.Track(pool.txFeed.Subscribe(ch))
|
||||
}
|
||||
|
||||
|
|
@ -436,7 +439,7 @@ func (self *TxPool) add(ctx context.Context, tx *types.Transaction) error {
|
|||
// Notify the subscribers. This event is posted in a goroutine
|
||||
// because it's possible that somewhere during the post "Remove transaction"
|
||||
// gets called which will then wait for the global tx pool lock and deadlock.
|
||||
go self.txFeed.Send(core.TxPreEvent{Tx: tx})
|
||||
go self.txFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
|
||||
}
|
||||
|
||||
// Print a log message if low enough level is set
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ const (
|
|||
resultQueueSize = 10
|
||||
miningLogAtDepth = 5
|
||||
|
||||
// txChanSize is the size of channel listening to TxPreEvent.
|
||||
// txChanSize is the size of channel listening to NewTxsEvent.
|
||||
// The number is referenced from the size of tx pool.
|
||||
txChanSize = 4096
|
||||
// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
|
||||
|
|
@ -109,8 +109,8 @@ type worker struct {
|
|||
|
||||
// update loop
|
||||
mux *event.TypeMux
|
||||
txCh chan core.TxPreEvent
|
||||
txSub event.Subscription
|
||||
txsCh chan core.NewTxsEvent
|
||||
txsSub event.Subscription
|
||||
chainHeadCh chan core.ChainHeadEvent
|
||||
chainHeadSub event.Subscription
|
||||
chainSideCh chan core.ChainSideEvent
|
||||
|
|
@ -149,7 +149,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
|
|||
engine: engine,
|
||||
eth: eth,
|
||||
mux: mux,
|
||||
txCh: make(chan core.TxPreEvent, txChanSize),
|
||||
txsCh: make(chan core.NewTxsEvent, txChanSize),
|
||||
chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize),
|
||||
chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize),
|
||||
chainDb: eth.ChainDb(),
|
||||
|
|
@ -163,8 +163,8 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
|
|||
announceTxs: announceTxs,
|
||||
}
|
||||
if worker.announceTxs {
|
||||
// Subscribe TxPreEvent for tx pool
|
||||
worker.txSub = eth.TxPool().SubscribeTxPreEvent(worker.txCh)
|
||||
// Subscribe NewTxsEvent for tx pool
|
||||
worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)
|
||||
}
|
||||
// Subscribe events for blockchain
|
||||
worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
|
||||
|
|
@ -261,7 +261,7 @@ func (self *worker) unregister(agent Agent) {
|
|||
|
||||
func (self *worker) update() {
|
||||
if self.announceTxs {
|
||||
defer self.txSub.Unsubscribe()
|
||||
defer self.txsSub.Unsubscribe()
|
||||
}
|
||||
defer self.chainHeadSub.Unsubscribe()
|
||||
defer self.chainSideSub.Unsubscribe()
|
||||
|
|
@ -314,13 +314,20 @@ func (self *worker) update() {
|
|||
self.uncleMu.Unlock()
|
||||
}
|
||||
|
||||
// Handle TxPreEvent
|
||||
case ev := <-self.txCh:
|
||||
// Apply transaction to the pending state if we're not mining
|
||||
// Handle NewTxsEvent
|
||||
case ev := <-self.txsCh:
|
||||
// Apply transactions to the pending state if we're not mining.
|
||||
//
|
||||
// Note all transactions received may not be continuous with transactions
|
||||
// already included in the current mining block. These transactions will
|
||||
// be automatically eliminated.
|
||||
if atomic.LoadInt32(&self.mining) == 0 {
|
||||
self.currentMu.Lock()
|
||||
acc, _ := types.Sender(self.current.signer, ev.Tx)
|
||||
txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
|
||||
txs := make(map[common.Address]types.Transactions)
|
||||
for _, tx := range ev.Txs {
|
||||
acc, _ := types.Sender(self.current.signer, tx)
|
||||
txs[acc] = append(txs[acc], tx)
|
||||
}
|
||||
feeCapacity := state.GetTRC21FeeCapacityFromState(self.current.state)
|
||||
txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs, nil, feeCapacity)
|
||||
self.current.commitTransactions(self.mux, feeCapacity, txset, specialTxs, self.chain, self.coinbase)
|
||||
|
|
|
|||
Loading…
Reference in a new issue