Skip to content

Commit bf5c07b

Browse files
libotonylibotony
authored
Merge pull request #245 from qianbin/tx-mark
Improve Tx Propagation
2 parents 3d1fe0a + 8f567ac commit bf5c07b

11 files changed

Lines changed: 87 additions & 55 deletions

File tree

‎cmd/thor/node/packer_loop.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616
"github.com/pkg/errors"
1717
"github.com/vechain/thor/packer"
1818
"github.com/vechain/thor/thor"
19+
"github.com/vechain/thor/tx"
1920
)
2021

2122
func (n *Node) packerLoop(ctx context.Context) {
@@ -83,10 +84,10 @@ func (n *Node) packerLoop(ctx context.Context) {
8384

8485
func (n *Node) pack(flow *packer.Flow) error {
8586
txs := n.txPool.Executables()
86-
var txsToRemove []thor.Bytes32
87+
var txsToRemove []*tx.Transaction
8788
defer func() {
88-
for _, id := range txsToRemove {
89-
n.txPool.Remove(id)
89+
for _, tx := range txsToRemove {
90+
n.txPool.Remove(tx.Hash(), tx.ID())
9091
}
9192
}()
9293

@@ -99,7 +100,7 @@ func (n *Node) pack(flow *packer.Flow) error {
99100
if packer.IsTxNotAdoptableNow(err) {
100101
continue
101102
}
102-
txsToRemove = append(txsToRemove, tx.ID())
103+
txsToRemove = append(txsToRemove, tx)
103104
}
104105
}
105106

‎cmd/thor/node/tx_stash.go‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package node
77

88
import (
9+
"bytes"
910
"container/list"
1011

1112
"github.com/ethereum/go-ethereum/rlp"
@@ -27,7 +28,7 @@ func newTxStash(kv kv.GetPutter, maxSize int) *txStash {
2728
}
2829

2930
func (ts *txStash) Save(tx *tx.Transaction) error {
30-
has, err := ts.kv.Has(tx.ID().Bytes())
31+
has, err := ts.kv.Has(tx.Hash().Bytes())
3132
if err != nil {
3233
return err
3334
}
@@ -40,10 +41,10 @@ func (ts *txStash) Save(tx *tx.Transaction) error {
4041
return err
4142
}
4243

43-
if err := ts.kv.Put(tx.ID().Bytes(), data); err != nil {
44+
if err := ts.kv.Put(tx.Hash().Bytes(), data); err != nil {
4445
return err
4546
}
46-
ts.fifo.PushBack(tx.ID())
47+
ts.fifo.PushBack(tx.Hash())
4748
for ts.fifo.Len() > ts.maxSize {
4849
keyToDelete := ts.fifo.Remove(ts.fifo.Front()).(thor.Bytes32).Bytes()
4950
if err := ts.kv.Delete(keyToDelete); err != nil {
@@ -54,6 +55,7 @@ func (ts *txStash) Save(tx *tx.Transaction) error {
5455
}
5556

5657
func (ts *txStash) LoadAll() tx.Transactions {
58+
batch := ts.kv.NewBatch()
5759
var txs tx.Transactions
5860
iter := ts.kv.NewIterator(*kv.NewRangeWithBytesPrefix(nil))
5961
for iter.Next() {
@@ -65,8 +67,19 @@ func (ts *txStash) LoadAll() tx.Transactions {
6567
}
6668
} else {
6769
txs = append(txs, &tx)
68-
ts.fifo.PushBack(tx.ID())
70+
ts.fifo.PushBack(tx.Hash())
71+
72+
// Keys were tx ids.
73+
// Here to remap values using tx hashes.
74+
if !bytes.Equal(iter.Key(), tx.Hash().Bytes()) {
75+
batch.Delete(iter.Key())
76+
batch.Put(tx.Hash().Bytes(), iter.Value())
77+
}
6978
}
7079
}
80+
81+
if err := batch.Write(); err != nil {
82+
log.Warn("remap stashed txs", "err", err)
83+
}
7184
return txs
7285
}

‎cmd/thor/solo/solo.go‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import (
2222
"github.com/vechain/thor/logdb"
2323
"github.com/vechain/thor/packer"
2424
"github.com/vechain/thor/state"
25-
"github.com/vechain/thor/thor"
2625
"github.com/vechain/thor/tx"
2726
"github.com/vechain/thor/txpool"
2827
)
@@ -118,10 +117,10 @@ func (s *Solo) loop(ctx context.Context) {
118117

119118
func (s *Solo) packing(pendingTxs tx.Transactions) error {
120119
best := s.chain.BestBlock()
121-
var txsToRemove []thor.Bytes32
120+
var txsToRemove []*tx.Transaction
122121
defer func() {
123-
for _, id := range txsToRemove {
124-
s.txPool.Remove(id)
122+
for _, tx := range txsToRemove {
123+
s.txPool.Remove(tx.Hash(), tx.ID())
125124
}
126125
}()
127126

@@ -142,7 +141,7 @@ func (s *Solo) packing(pendingTxs tx.Transactions) error {
142141
case packer.IsTxNotAdoptableNow(err):
143142
continue
144143
default:
145-
txsToRemove = append(txsToRemove, tx.ID())
144+
txsToRemove = append(txsToRemove, tx)
146145
}
147146
}
148147

‎comm/handle_rpc.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -69,8 +69,8 @@ func (c *Communicator) handleRPC(peer *Peer, msg *p2p.Msg, write func(interface{
6969
if err := msg.Decode(&newTx); err != nil {
7070
return errors.WithMessage(err, "decode msg")
7171
}
72-
peer.MarkTransaction(newTx.ID())
73-
c.txPool.StrictlyAdd(newTx)
72+
peer.MarkTransaction(newTx.Hash())
73+
c.txPool.Add(newTx)
7474
write(&struct{}{})
7575
case proto.MsgGetBlockByID:
7676
var blockID thor.Bytes32
@@ -146,10 +146,10 @@ func (c *Communicator) handleRPC(peer *Peer, msg *p2p.Msg, write func(interface{
146146

147147
for _, tx := range txsToSync.txs {
148148
n++
149-
if peer.IsTransactionKnown(tx.ID()) {
149+
if peer.IsTransactionKnown(tx.Hash()) {
150150
continue
151151
}
152-
peer.MarkTransaction(tx.ID())
152+
peer.MarkTransaction(tx.Hash())
153153
toSend = append(toSend, tx)
154154
size += tx.Size()
155155
if size >= maxTxSyncSize {

‎comm/peer.go‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,9 @@ import (
2020
)
2121

2222
const (
23-
maxKnownTxs = 32768 // Maximum transactions IDs to keep in the known list (prevent DOS)
24-
maxKnownBlocks = 1024 // Maximum block IDs to keep in the known list (prevent DOS)
23+
maxKnownTxs = 32768 // Maximum transactions IDs to keep in the known list (prevent DOS)
24+
maxKnownBlocks = 1024 // Maximum block IDs to keep in the known list (prevent DOS)
25+
knownTxMarkExpiration = 10 // Time in seconds to expire known tx mark
2526
)
2627

2728
func init() {
@@ -82,8 +83,8 @@ func (p *Peer) UpdateHead(id thor.Bytes32, totalScore uint64) {
8283
}
8384

8485
// MarkTransaction marks a transaction to known.
85-
func (p *Peer) MarkTransaction(id thor.Bytes32) {
86-
p.knownTxs.Add(id, struct{}{})
86+
func (p *Peer) MarkTransaction(hash thor.Bytes32) {
87+
p.knownTxs.Add(hash, time.Now().Unix())
8788
}
8889

8990
// MarkBlock marks a block to known.
@@ -92,8 +93,12 @@ func (p *Peer) MarkBlock(id thor.Bytes32) {
9293
}
9394

9495
// IsTransactionKnown returns if the transaction is known.
95-
func (p *Peer) IsTransactionKnown(id thor.Bytes32) bool {
96-
return p.knownTxs.Contains(id)
96+
func (p *Peer) IsTransactionKnown(hash thor.Bytes32) bool {
97+
ts, ok := p.knownTxs.Get(hash)
98+
if !ok {
99+
return false
100+
}
101+
return ts.(int64)+knownTxMarkExpiration > time.Now().Unix()
97102
}
98103

99104
// IsBlockKnown returns if the block is known.

‎comm/sync.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,7 +193,7 @@ func (c *Communicator) syncTxs(peer *Peer) {
193193
}
194194

195195
for _, tx := range result {
196-
peer.MarkTransaction(tx.ID())
196+
peer.MarkTransaction(tx.Hash())
197197
c.txPool.StrictlyAdd(tx)
198198
select {
199199
case <-c.ctx.Done():

‎comm/txs_loop.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,12 @@ func (c *Communicator) txsLoop() {
2424
if txEv.Executable != nil && *txEv.Executable {
2525
tx := txEv.Tx
2626
peers := c.peerSet.Slice().Filter(func(p *Peer) bool {
27-
return !p.IsTransactionKnown(tx.ID())
27+
return !p.IsTransactionKnown(tx.Hash())
2828
})
2929

3030
for _, peer := range peers {
3131
peer := peer
32-
peer.MarkTransaction(tx.ID())
32+
peer.MarkTransaction(tx.Hash())
3333
c.goes.Go(func() {
3434
if err := proto.NotifyNewTx(c.ctx, peer, tx); err != nil {
3535
peer.logger.Debug("failed to broadcast tx", "err", err)

‎tx/transaction.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ type Transaction struct {
3737
unprovedWork atomic.Value
3838
size atomic.Value
3939
intrinsicGas atomic.Value
40+
hash atomic.Value
4041
}
4142
}
4243

@@ -102,6 +103,19 @@ func (t *Transaction) ID() (id thor.Bytes32) {
102103
return
103104
}
104105

106+
// Hash returns hash of tx.
107+
// Unlike ID, it's the hash of RLP encoded tx.
108+
func (t *Transaction) Hash() (hash thor.Bytes32) {
109+
if cached := t.cache.hash.Load(); cached != nil {
110+
return cached.(thor.Bytes32)
111+
}
112+
defer func() { t.cache.hash.Store(hash) }()
113+
hw := thor.NewBlake2b()
114+
rlp.Encode(hw, t)
115+
hw.Sum(hash[:0])
116+
return
117+
}
118+
105119
// UnprovedWork returns unproved work of this tx.
106120
// It returns 0, if tx is not signed.
107121
func (t *Transaction) UnprovedWork() (w *big.Int) {

‎txpool/tx_object_map.go‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ import (
1313
"github.com/vechain/thor/tx"
1414
)
1515

16-
// txObjectMap to maintain mapping of ID to tx object, and account quota.
16+
// txObjectMap to maintain mapping of tx hash to tx object, and account quota.
1717
type txObjectMap struct {
1818
lock sync.RWMutex
1919
txObjMap map[thor.Bytes32]*txObject
@@ -27,18 +27,18 @@ func newTxObjectMap() *txObjectMap {
2727
}
2828
}
2929

30-
func (m *txObjectMap) Contains(txID thor.Bytes32) bool {
30+
func (m *txObjectMap) Contains(txHash thor.Bytes32) bool {
3131
m.lock.RLock()
3232
defer m.lock.RUnlock()
33-
_, found := m.txObjMap[txID]
33+
_, found := m.txObjMap[txHash]
3434
return found
3535
}
3636

3737
func (m *txObjectMap) Add(txObj *txObject, limitPerAccount int) error {
3838
m.lock.Lock()
3939
defer m.lock.Unlock()
4040

41-
if _, found := m.txObjMap[txObj.ID()]; found {
41+
if _, found := m.txObjMap[txObj.Hash()]; found {
4242
return nil
4343
}
4444

@@ -47,21 +47,21 @@ func (m *txObjectMap) Add(txObj *txObject, limitPerAccount int) error {
4747
}
4848

4949
m.quota[txObj.Origin()]++
50-
m.txObjMap[txObj.ID()] = txObj
50+
m.txObjMap[txObj.Hash()] = txObj
5151
return nil
5252
}
5353

54-
func (m *txObjectMap) Remove(txID thor.Bytes32) bool {
54+
func (m *txObjectMap) Remove(txHash thor.Bytes32) bool {
5555
m.lock.Lock()
5656
defer m.lock.Unlock()
5757

58-
if txObj, ok := m.txObjMap[txID]; ok {
58+
if txObj, ok := m.txObjMap[txHash]; ok {
5959
if m.quota[txObj.Origin()] > 1 {
6060
m.quota[txObj.Origin()]--
6161
} else {
6262
delete(m.quota, txObj.Origin())
6363
}
64-
delete(m.txObjMap, txID)
64+
delete(m.txObjMap, txHash)
6565
return true
6666
}
6767
return false
@@ -93,13 +93,13 @@ func (m *txObjectMap) Fill(txObjs []*txObject) {
9393
m.lock.Lock()
9494
defer m.lock.Unlock()
9595
for _, txObj := range txObjs {
96-
if _, found := m.txObjMap[txObj.ID()]; found {
96+
if _, found := m.txObjMap[txObj.Hash()]; found {
9797
continue
9898
}
9999
// skip account limit check
100100

101101
m.quota[txObj.Origin()]++
102-
m.txObjMap[txObj.ID()] = txObj
102+
m.txObjMap[txObj.Hash()] = txObj
103103
}
104104
}
105105

‎txpool/tx_object_map_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,13 +41,13 @@ func TestTxObjMap(t *testing.T) {
4141
assert.Nil(t, m.Add(txObj3, 1))
4242
assert.Equal(t, 2, m.Len())
4343

44-
assert.True(t, m.Contains(tx1.ID()))
45-
assert.False(t, m.Contains(tx2.ID()))
46-
assert.True(t, m.Contains(tx3.ID()))
44+
assert.True(t, m.Contains(tx1.Hash()))
45+
assert.False(t, m.Contains(tx2.Hash()))
46+
assert.True(t, m.Contains(tx3.Hash()))
4747

48-
assert.True(t, m.Remove(tx1.ID()))
49-
assert.False(t, m.Contains(tx1.ID()))
50-
assert.False(t, m.Remove(tx2.ID()))
48+
assert.True(t, m.Remove(tx1.Hash()))
49+
assert.False(t, m.Contains(tx1.Hash()))
50+
assert.False(t, m.Remove(tx2.Hash()))
5151

5252
assert.Equal(t, []*txObject{txObj3}, m.ToTxObjects())
5353
assert.Equal(t, tx.Transactions{tx3}, m.ToTxs())

0 commit comments

Comments
 (0)