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
|
package ethchain
import (
"bytes"
"container/list"
"errors"
"fmt"
"github.com/ethereum/eth-go/ethutil"
"github.com/ethereum/eth-go/ethwire"
"log"
"math/big"
"sync"
)
const (
txPoolQueueSize = 50
)
type TxPoolHook chan *Transaction
func FindTx(pool *list.List, finder func(*Transaction, *list.Element) bool) *Transaction {
for e := pool.Front(); e != nil; e = e.Next() {
if tx, ok := e.Value.(*Transaction); ok {
if finder(tx, e) {
return tx
}
}
}
return nil
}
type PublicSpeaker interface {
Broadcast(msgType ethwire.MsgType, data []interface{})
}
// The tx pool a thread safe transaction pool handler. In order to
// guarantee a non blocking pool we use a queue channel which can be
// independently read without needing access to the actual pool. If the
// pool is being drained or synced for whatever reason the transactions
// will simple queue up and handled when the mutex is freed.
type TxPool struct {
//server *Server
Speaker PublicSpeaker
// The mutex for accessing the Tx pool.
mutex sync.Mutex
// Queueing channel for reading and writing incoming
// transactions to
queueChan chan *Transaction
// Quiting channel
quit chan bool
// The actual pool
pool *list.List
BlockManager *BlockManager
Hook TxPoolHook
}
func NewTxPool() *TxPool {
return &TxPool{
//server: s,
mutex: sync.Mutex{},
pool: list.New(),
queueChan: make(chan *Transaction, txPoolQueueSize),
quit: make(chan bool),
}
}
// Blocking function. Don't use directly. Use QueueTransaction instead
func (pool *TxPool) addTransaction(tx *Transaction) {
pool.mutex.Lock()
pool.pool.PushBack(tx)
pool.mutex.Unlock()
// Broadcast the transaction to the rest of the peers
pool.Speaker.Broadcast(ethwire.MsgTxTy, []interface{}{tx.RlpData()})
}
// Process transaction validates the Tx and processes funds from the
// sender to the recipient.
func (pool *TxPool) ProcessTransaction(tx *Transaction, block *Block) (err error) {
log.Printf("[TXPL] Processing Tx %x\n", tx.Hash())
defer func() {
if r := recover(); r != nil {
log.Println(r)
err = fmt.Errorf("%v", r)
}
}()
// Get the sender
sender := block.GetAddr(tx.Sender())
// Make sure there's enough in the sender's account. Having insufficient
// funds won't invalidate this transaction but simple ignores it.
totAmount := new(big.Int).Add(tx.Value, new(big.Int).Mul(TxFee, TxFeeRat))
if sender.Amount.Cmp(totAmount) < 0 {
return errors.New("Insufficient amount in sender's account")
}
if sender.Nonce != tx.Nonce {
if ethutil.Config.Debug {
return fmt.Errorf("Invalid nonce %d(%d) continueing anyway", tx.Nonce, sender.Nonce)
} else {
return fmt.Errorf("Invalid nonce %d(%d)", tx.Nonce, sender.Nonce)
}
}
// Subtract the amount from the senders account
sender.Amount.Sub(sender.Amount, totAmount)
sender.Nonce += 1
// Get the receiver
receiver := block.GetAddr(tx.Recipient)
// Add the amount to receivers account which should conclude this transaction
receiver.Amount.Add(receiver.Amount, tx.Value)
block.UpdateAddr(tx.Sender(), sender)
block.UpdateAddr(tx.Recipient, receiver)
return
}
func (pool *TxPool) ValidateTransaction(tx *Transaction) error {
// Get the last block so we can retrieve the sender and receiver from
// the merkle trie
block := pool.BlockManager.BlockChain().CurrentBlock
// Something has gone horribly wrong if this happens
if block == nil {
return errors.New("No last block on the block chain")
}
// Get the sender
sender := block.GetAddr(tx.Sender())
totAmount := new(big.Int).Add(tx.Value, new(big.Int).Mul(TxFee, TxFeeRat))
// Make sure there's enough in the sender's account. Having insufficient
// funds won't invalidate this transaction but simple ignores it.
if sender.Amount.Cmp(totAmount) < 0 {
return fmt.Errorf("Insufficient amount in sender's (%x) account", tx.Sender())
}
// Increment the nonce making each tx valid only once to prevent replay
// attacks
return nil
}
func (pool *TxPool) queueHandler() {
out:
for {
select {
case tx := <-pool.queueChan:
hash := tx.Hash()
foundTx := FindTx(pool.pool, func(tx *Transaction, e *list.Element) bool {
return bytes.Compare(tx.Hash(), hash) == 0
})
if foundTx != nil {
break
}
// Validate the transaction
err := pool.ValidateTransaction(tx)
if err != nil {
if ethutil.Config.Debug {
log.Println("Validating Tx failed", err)
}
} else {
// Call blocking version. At this point it
// doesn't matter since this is a goroutine
pool.addTransaction(tx)
if pool.Hook != nil {
pool.Hook <- tx
}
}
case <-pool.quit:
break out
}
}
}
func (pool *TxPool) QueueTransaction(tx *Transaction) {
pool.queueChan <- tx
}
func (pool *TxPool) Flush() []*Transaction {
pool.mutex.Lock()
defer pool.mutex.Unlock()
txList := make([]*Transaction, pool.pool.Len())
i := 0
for e := pool.pool.Front(); e != nil; e = e.Next() {
if tx, ok := e.Value.(*Transaction); ok {
txList[i] = tx
}
i++
}
// Recreate a new list all together
// XXX Is this the fastest way?
pool.pool = list.New()
return txList
}
func (pool *TxPool) Start() {
go pool.queueHandler()
}
func (pool *TxPool) Stop() {
log.Println("[TXP] Stopping...")
close(pool.quit)
pool.Flush()
}
|
