-
Notifications
You must be signed in to change notification settings - Fork 113
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #4681 from oasisprotocol/pro-wh/feature/txunion
go/runtime/txpool: transactions in incoming messages
- Loading branch information
Showing
19 changed files
with
774 additions
and
297 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| go/runtime/txpool: Add roothash incoming messages' data as transactions | ||
|
|
||
| Roothash incoming messages can provide a piece of data for the runtime. | ||
| With this change, the data is now treated as a transaction. | ||
|
|
||
| Along with this change, we're splitting the txpool into multiple queues. | ||
| The transactions collected from roothash incoming messages go in a special | ||
| queue that does not undergo checking or broadcasting. | ||
|
|
||
| We also make another queue for a node's own transactions, so that a proposer | ||
| can prioritize its own transactions. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package txpool | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/oasisprotocol/oasis-core/go/common/crypto/hash" | ||
| "github.com/oasisprotocol/oasis-core/go/runtime/host/protocol" | ||
| ) | ||
|
|
||
| var ( | ||
| _ UsableTransactionSource = (*localQueue)(nil) | ||
| _ RecheckableTransactionStore = (*localQueue)(nil) | ||
| _ RepublishableTransactionSource = (*localQueue)(nil) | ||
| ) | ||
|
|
||
| // localQueue is a "front of the line" area for txs from our own node. We also keep these txs in order. | ||
| type localQueue struct { | ||
| l sync.Mutex | ||
| txs []*TxQueueMeta | ||
| indexesByHash map[hash.Hash]int | ||
| } | ||
|
|
||
| func newLocalQueue() *localQueue { | ||
| return &localQueue{ | ||
| indexesByHash: map[hash.Hash]int{}, | ||
| } | ||
| } | ||
|
|
||
| func (lq *localQueue) GetSchedulingSuggestion(countHint uint32) []*TxQueueMeta { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| return append([]*TxQueueMeta(nil), lq.txs...) | ||
| } | ||
|
|
||
| func (lq *localQueue) GetTxByHash(h hash.Hash) *TxQueueMeta { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| i, ok := lq.indexesByHash[h] | ||
| if !ok { | ||
| return nil | ||
| } | ||
| return lq.txs[i] | ||
| } | ||
|
|
||
| func (lq *localQueue) HandleTxsUsed(hashes []hash.Hash) { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| origCount := len(lq.txs) | ||
| keptCount := origCount | ||
| for _, h := range hashes { | ||
| if i, ok := lq.indexesByHash[h]; ok { | ||
| delete(lq.indexesByHash, h) | ||
| lq.txs[i] = nil | ||
| keptCount-- | ||
| } | ||
| } | ||
| if keptCount == origCount { | ||
| return | ||
| } | ||
| keptTxs := make([]*TxQueueMeta, 0, keptCount) | ||
| for _, tx := range lq.txs { | ||
| if tx == nil { | ||
| continue | ||
| } | ||
| i := len(keptTxs) | ||
| keptTxs = append(keptTxs, tx) | ||
| lq.indexesByHash[tx.Hash()] = i | ||
| } | ||
| lq.txs = keptTxs | ||
| } | ||
|
|
||
| func (lq *localQueue) TakeAll() []*TxQueueMeta { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| txs := lq.txs | ||
| lq.txs = nil | ||
| lq.indexesByHash = make(map[hash.Hash]int) | ||
| return txs | ||
| } | ||
|
|
||
| func (lq *localQueue) OfferChecked(tx *TxQueueMeta, _ *protocol.CheckTxMetadata) error { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| lq.indexesByHash[tx.Hash()] = len(lq.txs) | ||
| lq.txs = append(lq.txs, tx) | ||
| return nil | ||
| } | ||
|
|
||
| func (lq *localQueue) GetTxsToPublish() []*TxQueueMeta { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| return append([]*TxQueueMeta(nil), lq.txs...) | ||
| } | ||
|
|
||
| func (lq *localQueue) size() int { | ||
| lq.l.Lock() | ||
| defer lq.l.Unlock() | ||
| return len(lq.txs) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package txpool | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/oasisprotocol/oasis-core/go/common/crypto/hash" | ||
| "github.com/oasisprotocol/oasis-core/go/runtime/host/protocol" | ||
| ) | ||
|
|
||
| func TestLocalQueueBasic(t *testing.T) { | ||
| lq := newLocalQueue() | ||
|
|
||
| require.Len(t, lq.GetSchedulingSuggestion(50), 0, "get scheduling suggestion") | ||
|
|
||
| // Add two transactions, with a higher priority one coming later. | ||
| rawA := []byte("a") | ||
| txA := &TxQueueMeta{raw: rawA, hash: hash.NewFromBytes(rawA)} | ||
| require.NoError(t, lq.OfferChecked(txA, &protocol.CheckTxMetadata{Priority: 1}), "offer checked a") | ||
| rawB := []byte("b") | ||
| txB := &TxQueueMeta{raw: rawB, hash: hash.NewFromBytes(rawB)} | ||
| require.NoError(t, lq.OfferChecked(txB, &protocol.CheckTxMetadata{Priority: 5}), "offer checked a") | ||
| require.Equal(t, 2, lq.size()) | ||
|
|
||
| // We should preserve the order. Publish in original order. | ||
| require.EqualValues(t, []*TxQueueMeta{txA, txB}, lq.GetTxsToPublish(), "get txs to publish") | ||
| // Schedule in original order. | ||
| require.EqualValues(t, []*TxQueueMeta{txA, txB}, lq.GetSchedulingSuggestion(50), "get scheduling suggestion") | ||
|
|
||
| tx := lq.GetTxByHash(txA.Hash()) | ||
| require.EqualValues(t, txA, tx, "get tx by hash a") | ||
| hashC := hash.NewFromBytes([]byte("c")) | ||
| tx = lq.GetTxByHash(hashC) | ||
| require.Nil(t, tx, "get tx by hash c") | ||
|
|
||
| lq.HandleTxsUsed([]hash.Hash{hashC}) | ||
| require.EqualValues(t, map[hash.Hash]int{txA.Hash(): 0, txB.Hash(): 1}, lq.indexesByHash, "after handle txs used absent") | ||
| lq.HandleTxsUsed([]hash.Hash{txA.Hash()}) | ||
| require.EqualValues(t, map[hash.Hash]int{txB.Hash(): 0}, lq.indexesByHash, "after handle txs used") | ||
|
|
||
| require.EqualValues(t, []*TxQueueMeta{txB}, lq.TakeAll(), "take all") | ||
| require.Len(t, lq.GetSchedulingSuggestion(50), 0, "after take all") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| package txpool | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/oasisprotocol/oasis-core/go/common/crypto/hash" | ||
| "github.com/oasisprotocol/oasis-core/go/runtime/host/protocol" | ||
| ) | ||
|
|
||
| var ( | ||
| _ UsableTransactionSource = (*mainQueue)(nil) | ||
| _ RecheckableTransactionStore = (*mainQueue)(nil) | ||
| _ RepublishableTransactionSource = (*mainQueue)(nil) | ||
| ) | ||
|
|
||
| // MainQueueTransaction is a transaction and its metadata in the main queue. | ||
| type MainQueueTransaction struct { | ||
| TxQueueMeta | ||
|
|
||
| // priority defines the transaction's priority as specified by the runtime. | ||
| priority uint64 | ||
|
|
||
| // sender is a unique transaction sender identifier as specified by the runtime. | ||
| sender string | ||
| // senderSeq is a per-sender sequence number as specified by the runtime. | ||
| senderSeq uint64 | ||
| } | ||
|
|
||
| func newTransaction(tx TxQueueMeta) *MainQueueTransaction { | ||
| return &MainQueueTransaction{ | ||
| TxQueueMeta: tx, | ||
| } | ||
| } | ||
|
|
||
| // String returns a string representation of a transaction. | ||
| func (tx *MainQueueTransaction) String() string { | ||
| return fmt.Sprintf("MainQueueTransaction{hash: %s, first_seen: %s, priority: %d}", tx.Hash(), tx.FirstSeen(), tx.priority) | ||
| } | ||
|
|
||
| // Priority returns the transaction priority. | ||
| func (tx *MainQueueTransaction) Priority() uint64 { | ||
| return tx.priority | ||
| } | ||
|
|
||
| // Sender returns the transaction sender. | ||
| func (tx *MainQueueTransaction) Sender() string { | ||
| return tx.sender | ||
| } | ||
|
|
||
| // SenderSeq returns the per-sender sequence number. | ||
| func (tx *MainQueueTransaction) SenderSeq() uint64 { | ||
| return tx.senderSeq | ||
| } | ||
|
|
||
| // setChecked populates transaction data retrieved from checks. | ||
| func (tx *MainQueueTransaction) setChecked(meta *protocol.CheckTxMetadata) { | ||
| if meta != nil { | ||
| tx.priority = meta.Priority | ||
| tx.sender = string(meta.Sender) | ||
| tx.senderSeq = meta.SenderSeq | ||
| } | ||
|
|
||
| // If the sender is empty (e.g. because the runtime does not support specifying a sender), we | ||
| // treat each transaction as having a unique sender. This is to allow backwards compatibility. | ||
| if len(tx.sender) == 0 { | ||
| h := tx.Hash() | ||
| tx.sender = string(h[:]) | ||
| } | ||
| } | ||
|
|
||
| // mainQueue is a priority queue for transactions that we give no special treatment. | ||
| type mainQueue struct { | ||
| // This implementation adapts the existing scheduleQueue code. | ||
| inner *scheduleQueue | ||
| } | ||
|
|
||
| func newMainQueue(capacity int) *mainQueue { | ||
| return &mainQueue{ | ||
| inner: newScheduleQueue(capacity), | ||
| } | ||
| } | ||
|
|
||
| func (mq *mainQueue) GetSchedulingSuggestion(countHint uint32) []*TxQueueMeta { | ||
| txMetas := mq.inner.getPrioritizedBatch(nil, countHint) | ||
| var txs []*TxQueueMeta | ||
| for _, txMeta := range txMetas { | ||
| txs = append(txs, &txMeta.TxQueueMeta) | ||
| } | ||
| return txs | ||
| } | ||
|
|
||
| func (mq *mainQueue) GetTxByHash(h hash.Hash) *TxQueueMeta { | ||
| txMetas, _ := mq.inner.getKnownBatch([]hash.Hash{h}) | ||
| if txMetas[0] == nil { | ||
| return nil | ||
| } | ||
| return &txMetas[0].TxQueueMeta | ||
| } | ||
|
|
||
| func (mq *mainQueue) HandleTxsUsed(hashes []hash.Hash) { | ||
| mq.inner.remove(hashes) | ||
| } | ||
|
|
||
| func (mq *mainQueue) GetSchedulingExtra(offset *hash.Hash, limit uint32) []*TxQueueMeta { | ||
| txMetas := mq.inner.getPrioritizedBatch(offset, limit) | ||
| var txs []*TxQueueMeta | ||
| for _, txMeta := range txMetas { | ||
| txs = append(txs, &txMeta.TxQueueMeta) | ||
| } | ||
| return txs | ||
| } | ||
|
|
||
| func (mq *mainQueue) TakeAll() []*TxQueueMeta { | ||
| txMetas := mq.inner.getAll() | ||
| mq.inner.clear() | ||
| var txs []*TxQueueMeta | ||
| for _, txMeta := range txMetas { | ||
| txs = append(txs, &txMeta.TxQueueMeta) | ||
| } | ||
| return txs | ||
| } | ||
|
|
||
| func (mq *mainQueue) OfferChecked(tx *TxQueueMeta, meta *protocol.CheckTxMetadata) error { | ||
| txMeta := newTransaction(*tx) | ||
| txMeta.setChecked(meta) | ||
|
|
||
| return mq.inner.add(txMeta) | ||
| } | ||
|
|
||
| func (mq *mainQueue) GetTxsToPublish() []*TxQueueMeta { | ||
| txMetas := mq.inner.getAll() | ||
| var txs []*TxQueueMeta | ||
| for _, txMeta := range txMetas { | ||
| txs = append(txs, &txMeta.TxQueueMeta) | ||
| } | ||
| return txs | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.