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
|
package main
import (
"container/list"
"errors"
"github.com/ethereum/ethutil-go"
"log"
"math/big"
"sync"
)
const (
txPoolQueueSize = 50
)
func FindTx(pool *list.List, finder func(*ethutil.Transaction, *list.Element) bool) *ethutil.Transaction {
for e := pool.Front(); e != nil; e = e.Next() {
if tx, ok := e.Value.(*ethutil.Transaction); ok {
if finder(tx, e) {
return tx
}
}
}
return nil
}
// 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
// The mutex for accessing the Tx pool.
mutex sync.Mutex
// Queueing channel for reading and writing incoming
// transactions to
queueChan chan *ethutil.Transaction
// Quiting channel
quit chan bool
pool *list.List
}
func NewTxPool(s *Server) *TxPool {
return &TxPool{
server: s,
mutex: sync.Mutex{},
pool: list.New(),
queueChan: make(chan *ethutil.Transaction, txPoolQueueSize),
quit: make(chan bool),
}
}
// Blocking function. Don't use directly. Use QueueTransaction instead
func (pool *TxPool) addTransaction(tx *ethutil.Transaction) {
pool.mutex.Lock()
defer pool.mutex.Unlock()
pool.pool.PushBack(tx)
}
// Process transaction validates the Tx and processes funds from the
// sender to the recipient.
func (pool *TxPool) processTransaction(tx *ethutil.Transaction) error {
// Get the last block so we can retrieve the sender and receiver from
// the merkle trie
block := pool.server.blockManager.bc.LastBlock
// Something has gone horribly wrong if this happens
if block == nil {
return errors.New("No last block on the block chain")
}
var sender, receiver *ethutil.Ether
// Get the sender
data := block.State().Get(string(tx.Sender()))
// If it doesn't exist create a new account. Of course trying to send funds
// from this account will fail since it will hold 0 Wei
if data == "" {
sender = ethutil.NewEther(big.NewInt(0))
} else {
sender = ethutil.NewEtherFromData([]byte(data))
}
// Defer the update. Whatever happens it should be persisted
defer block.State().Update(string(tx.Sender()), string(sender.RlpEncode()))
// 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(tx.Value) < 0 {
return errors.New("Insufficient amount in sender's account")
}
// Subtract the amount from the senders account
sender.Amount.Sub(sender.Amount, tx.Value)
// Increment the nonce making each tx valid only once to prevent replay
// attacks
sender.Nonce += 1
// Get the receiver
data = block.State().Get(tx.Recipient)
// If the receiver doesn't exist yet, create a new account to which the
// funds will be send.
if data == "" {
receiver = ethutil.NewEther(big.NewInt(0))
} else {
receiver = ethutil.NewEtherFromData([]byte(data))
}
// Defer the update
defer block.State().Update(tx.Recipient, string(receiver.RlpEncode()))
// Add the amount to receivers account which should conclude this transaction
receiver.Amount.Add(receiver.Amount, tx.Value)
return nil
}
func (pool *TxPool) queueHandler() {
out:
for {
select {
case tx := <-pool.queueChan:
// Process the transaction
err := pool.processTransaction(tx)
if err != nil {
log.Println("Error processing Tx", err)
} else {
// Call blocking version. At this point it
// doesn't matter since this is a goroutine
pool.addTransaction(tx)
}
case <-pool.quit:
break out
}
}
}
func (pool *TxPool) QueueTransaction(tx *ethutil.Transaction) {
pool.queueChan <- tx
}
func (pool *TxPool) Flush() {
pool.mutex.Lock()
defer pool.mutex.Unlock()
pool.mutex.Unlock()
}
func (pool *TxPool) Start() {
go pool.queueHandler()
}
func (pool *TxPool) Stop() {
log.Println("[TXP] Stopping...")
close(pool.quit)
pool.Flush()
}
|
