aboutsummaryrefslogtreecommitdiffstats
path: root/core/transaction_pool.go
blob: e31f5c6b3ea980b5dc4dc04031582b3aad632811 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
package core

import (
    "errors"
    "fmt"
    "math/big"
    "sort"
    "sync"

    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/core/state"
    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/event"
    "github.com/ethereum/go-ethereum/logger"
    "github.com/ethereum/go-ethereum/logger/glog"
)

var (
    // Transaction Pool Errors
    ErrInvalidSender      = errors.New("Invalid sender")
    ErrNonce              = errors.New("Nonce too low")
    ErrCheap              = errors.New("Gas price too low for acceptance")
    ErrBalance            = errors.New("Insufficient balance")
    ErrNonExistentAccount = errors.New("Account does not exist or account balance too low")
    ErrInsufficientFunds  = errors.New("Insufficient funds for gas * price + value")
    ErrIntrinsicGas       = errors.New("Intrinsic gas too low")
    ErrGasLimit           = errors.New("Exceeds block gas limit")
    ErrNegativeValue      = errors.New("Negative value")
)

const (
    maxQueued = 200 // max limit of queued txs per address
)

type stateFn func() *state.StateDB

// TxPool contains all currently known transactions. Transactions
// enter the pool when they are received from the network or submitted
// locally. They exit the pool when they are included in the blockchain.
//
// The pool separates processable transactions (which can be applied to the
// current state) and future transactions. Transactions move between those
// two states over time as they are received and processed.
type TxPool struct {
    quit         chan bool // Quiting channel
    currentState stateFn   // The state function which will allow us to do some pre checkes
    pendingState *state.ManagedState
    gasLimit     func() *big.Int // The current gas limit function callback
    minGasPrice  *big.Int
    eventMux     *event.TypeMux
    events       event.Subscription

    mu      sync.RWMutex
    pending map[common.Hash]*types.Transaction // processable transactions
    queue   map[common.Address]map[common.Hash]*types.Transaction
}

func NewTxPool(eventMux *event.TypeMux, currentStateFn stateFn, gasLimitFn func() *big.Int) *TxPool {
    pool := &TxPool{
        pending:      make(map[common.Hash]*types.Transaction),
        queue:        make(map[common.Address]map[common.Hash]*types.Transaction),
        quit:         make(chan bool),
        eventMux:     eventMux,
        currentState: currentStateFn,
        gasLimit:     gasLimitFn,
        minGasPrice:  new(big.Int),
        pendingState: state.ManageState(currentStateFn()),
        events:       eventMux.Subscribe(ChainEvent{}, GasPriceChanged{}),
    }
    go pool.eventLoop()

    return pool
}

func (pool *TxPool) eventLoop() {
    // Track chain events. When a chain events occurs (new chain canon block)
    // we need to know the new state. The new state will help us determine
    // the nonces in the managed state
    for ev := range pool.events.Chan() {
        pool.mu.Lock()

        switch ev := ev.(type) {
        case ChainEvent:
            pool.resetState()
        case GasPriceChanged:
            pool.minGasPrice = ev.Price
        }

        pool.mu.Unlock()
    }
}

func (pool *TxPool) resetState() {
    pool.pendingState = state.ManageState(pool.currentState())

    // validate the pool of pending transactions, this will remove
    // any transactions that have been included in the block or
    // have been invalidated because of another transaction (e.g.
    // higher gas price)
    pool.validatePool()

    // Loop over the pending transactions and base the nonce of the new
    // pending transaction set.
    for _, tx := range pool.pending {
        if addr, err := tx.From(); err == nil {
            // Set the nonce. Transaction nonce can never be lower
            // than the state nonce; validatePool took care of that.
            pool.pendingState.SetNonce(addr, tx.Nonce())
        }
    }

    // Check the queue and move transactions over to the pending if possible
    // or remove those that have become invalid
    pool.checkQueue()
}

func (pool *TxPool) Stop() {
    close(pool.quit)
    pool.events.Unsubscribe()
    glog.V(logger.Info).Infoln("TX Pool stopped")
}

func (pool *TxPool) State() *state.ManagedState {
    pool.mu.RLock()
    defer pool.mu.RUnlock()

    return pool.pendingState
}

// validateTx checks whether a transaction is valid according
// to the consensus rules.
func (pool *TxPool) validateTx(tx *types.Transaction) error {
    // Validate sender
    var (
        from common.Address
        err  error
    )

    // Drop transactions under our own minimal accepted gas price
    if pool.minGasPrice.Cmp(tx.GasPrice()) > 0 {
        return ErrCheap
    }

    // Validate the transaction sender and it's sig. Throw
    // if the from fields is invalid.
    if from, err = tx.From(); err != nil {
        return ErrInvalidSender
    }

    // Make sure the account exist. Non existant accounts
    // haven't got funds and well therefor never pass.
    if !pool.currentState().HasAccount(from) {
        return ErrNonExistentAccount
    }

    // Check the transaction doesn't exceed the current
    // block limit gas.
    if pool.gasLimit().Cmp(tx.GasLimit) < 0 {
        return ErrGasLimit
    }

    // Transactions can't be negative. This may never happen
    // using RLP decoded transactions but may occur if you create
    // a transaction using the RPC for example.
    if tx.Amount.Cmp(common.Big0) < 0 {
        return ErrNegativeValue
    }

    // Transactor should have enough funds to cover the costs
    // cost == V + GP * GL
    total := new(big.Int).Mul(tx.Price, tx.GasLimit)
    total.Add(total, tx.Value())
    if pool.currentState().GetBalance(from).Cmp(total) < 0 {
        return ErrInsufficientFunds
    }

    // Should supply enough intrinsic gas
    if tx.GasLimit.Cmp(IntrinsicGas(tx)) < 0 {
        return ErrIntrinsicGas
    }

    // Last but not least check for nonce errors (intensive
    // operation, saved for last)
    if pool.currentState().GetNonce(from) > tx.Nonce() {
        return ErrNonce
    }

    return nil
}

// validate and queue transactions.
func (self *TxPool) add(tx *types.Transaction) error {
    hash := tx.Hash()

    if self.pending[hash] != nil {
        return fmt.Errorf("Known transaction (%x)", hash[:4])
    }
    err := self.validateTx(tx)
    if err != nil {
        return err
    }
    self.queueTx(hash, tx)

    if glog.V(logger.Debug) {
        var toname string
        if to := tx.To(); to != nil {
            toname = common.Bytes2Hex(to[:4])
        } else {
            toname = "[NEW_CONTRACT]"
        }
        // we can ignore the error here because From is
        // verified in ValidateTransaction.
        f, _ := tx.From()
        from := common.Bytes2Hex(f[:4])
        glog.Infof("(t) %x => %s (%v) %x\n", from, toname, tx.Value, hash)
    }

    // check and validate the queueue
    self.checkQueue()

    return nil
}

// queueTx will queue an unknown transaction
func (self *TxPool) queueTx(hash common.Hash, tx *types.Transaction) {
    from, _ := tx.From() // already validated
    if self.queue[from] == nil {
        self.queue[from] = make(map[common.Hash]*types.Transaction)
    }
    self.queue[from][hash] = tx
}

// addTx will add a transaction to the pending (processable queue) list of transactions
func (pool *TxPool) addTx(hash common.Hash, addr common.Address, tx *types.Transaction) {
    if _, ok := pool.pending[hash]; !ok {
        pool.pending[hash] = tx

        // Increment the nonce on the pending state. This can only happen if
        // the nonce is +1 to the previous one.
        pool.pendingState.SetNonce(addr, tx.AccountNonce+1)
        // 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 pool.eventMux.Post(TxPreEvent{tx})
    }
}

// Add queues a single transaction in the pool if it is valid.
func (self *TxPool) Add(tx *types.Transaction) error {
    self.mu.Lock()
    defer self.mu.Unlock()

    return self.add(tx)
}

// AddTransactions attempts to queue all valid transactions in txs.
func (self *TxPool) AddTransactions(txs []*types.Transaction) {
    self.mu.Lock()
    defer self.mu.Unlock()

    for _, tx := range txs {
        if err := self.add(tx); err != nil {
            glog.V(logger.Debug).Infoln("tx error:", err)
        } else {
            h := tx.Hash()
            glog.V(logger.Debug).Infof("tx %x\n", h[:4])
        }
    }
}

// GetTransaction returns a transaction if it is contained in the pool
// and nil otherwise.
func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
    // check the txs first
    if tx, ok := tp.pending[hash]; ok {
        return tx
    }
    // check queue
    for _, txs := range tp.queue {
        if tx, ok := txs[hash]; ok {
            return tx
        }
    }
    return nil
}

// GetTransactions returns all currently processable transactions.
// The returned slice may be modified by the caller.
func (self *TxPool) GetTransactions() (txs types.Transactions) {
    self.mu.Lock()
    defer self.mu.Unlock()

    // check queue first
    self.checkQueue()
    // invalidate any txs
    self.validatePool()

    txs = make(types.Transactions, len(self.pending))
    i := 0
    for _, tx := range self.pending {
        txs[i] = tx
        i++
    }
    return txs
}

// GetQueuedTransactions returns all non-processable transactions.
func (self *TxPool) GetQueuedTransactions() types.Transactions {
    self.mu.RLock()
    defer self.mu.RUnlock()

    var ret types.Transactions
    for _, txs := range self.queue {
        for _, tx := range txs {
            ret = append(ret, tx)
        }
    }
    sort.Sort(types.TxByNonce{ret})
    return ret
}

// RemoveTransactions removes all given transactions from the pool.
func (self *TxPool) RemoveTransactions(txs types.Transactions) {
    self.mu.Lock()
    defer self.mu.Unlock()
    for _, tx := range txs {
        self.removeTx(tx.Hash())
    }
}

// checkQueue moves transactions that have become processable to main pool.
func (pool *TxPool) checkQueue() {
    state := pool.pendingState

    var addq txQueue
    for address, txs := range pool.queue {
        // guessed nonce is the nonce currently kept by the tx pool (pending state)
        guessedNonce := state.GetNonce(address)
        // true nonce is the nonce known by the last state
        trueNonce := pool.currentState().GetNonce(address)
        addq := addq[:0]
        for hash, tx := range txs {
            if tx.AccountNonce < trueNonce {
                // Drop queued transactions whose nonce is lower than
                // the account nonce because they have been processed.
                delete(txs, hash)
            } else {
                // Collect the remaining transactions for the next pass.
                addq = append(addq, txQueueEntry{hash, address, tx})
            }
        }
        // Find the next consecutive nonce range starting at the
        // current account nonce.
        sort.Sort(addq)
        for i, e := range addq {
            // start deleting the transactions from the queue if they exceed the limit
            if i > maxQueued {
                if glog.V(logger.Debug) {
                    glog.Infof("Queued tx limit exceeded for %s. Tx %s removed\n", common.PP(address[:]), common.PP(e.hash[:]))
                }
                delete(pool.queue[address], e.hash)
                continue
            }

            if e.AccountNonce > guessedNonce {
                break
            }
            delete(txs, e.hash)
            pool.addTx(e.hash, address, e.Transaction)
        }
        // Delete the entire queue entry if it became empty.
        if len(txs) == 0 {
            delete(pool.queue, address)
        }
    }
}

func (pool *TxPool) removeTx(hash common.Hash) {
    // delete from pending pool
    delete(pool.pending, hash)
    // delete from queue
    for address, txs := range pool.queue {
        if _, ok := txs[hash]; ok {
            if len(txs) == 1 {
                // if only one tx, remove entire address entry.
                delete(pool.queue, address)
            } else {
                delete(txs, hash)
            }
            break
        }
    }
}

// validatePool removes invalid and processed transactions from the main pool.
func (pool *TxPool) validatePool() {
    for hash, tx := range pool.pending {
        if err := pool.validateTx(tx); err != nil {
            if glog.V(logger.Core) {
                glog.Infof("removed tx (%x) from pool: %v\n", hash[:4], err)
            }
            delete(pool.pending, hash)
        }
    }
}

type txQueue []txQueueEntry

type txQueueEntry struct {
    hash common.Hash
    addr common.Address
    *types.Transaction
}

func (q txQueue) Len() int           { return len(q) }
func (q txQueue) Swap(i, j int)      { q[i], q[j] = q[j], q[i] }
func (q txQueue) Less(i, j int) bool { return q[i].AccountNonce < q[j].AccountNonce }