Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Mempool concurrency: transaction validation #103

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Mempool concurrency: transaction validation #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4847990
comment runtimes; parallelize transaction verification in mempool
kirushanths Feb 23, 2019
c72f023
Merge branch 'master' of https://github.com/kyokan/plasma into ks-mem…
kirushanths Feb 24, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pkg/eth/signatures.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,22 +21,27 @@ func Sign(privKey *ecdsa.PrivateKey, hash util.Hash) (chain.Signature, error) {
}

func ValidateSignature(hash, signature []byte, address common.Address) error {
// 2 Us
ethHash := util.GethHash(hash)

// 200 ns
sigCopy := make([]byte, len(signature))
copy(sigCopy, signature)
if len(sigCopy) == 65 && sigCopy[64] > 26 {
sigCopy[64] -= 27
}

// 220 Us
pubKey, err := crypto.SigToPub(ethHash, sigCopy)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Basically 70% of the work happens right here on this line.

Copy link
Contributor Author

@kirushanths kirushanths Feb 24, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you do the rough math here: If this takes say on average 200 us. That means about 5 transactions processed a millisecond. And in a 100ms block with 16-cores (my setup) parallelizing the processing you get: 5 transactions/ms x 100 ms/block x 16cores = 8000 TPS. I'm basically this figure plus/minus overhead.

EDIT: Sorry my math was horrible. And this line doesn't average near 200us. It's much lower but it's still the main bottleneck during transaction validation and eating 70% of the time. This is pre-block-packaging.

if err != nil {
return err
}

// 3 Us
signatureAddress := crypto.PubkeyToAddress(*pubKey)
if address != signatureAddress {
return errors.New("invalid signature")
}

return nil
}
78 changes: 48 additions & 30 deletions pkg/service/mempool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ type TxInclusionResponse struct {

type Mempool struct {
txReqs chan *txRequest
validatedTxReqs chan *txRequest
quit chan bool
flushReq chan flushSpendReq
txPool []MempoolTx
Expand All @@ -50,36 +51,58 @@ type flushSpendReq struct {

func NewMempool(storage db.Storage, client eth.Client) *Mempool {
return &Mempool{
txReqs: make(chan *txRequest),
quit: make(chan bool),
flushReq: make(chan flushSpendReq),
txPool: make([]MempoolTx, 0),
poolSpends: make(map[string]bool),
storage: storage,
client: client,
txReqs: make(chan *txRequest),
validatedTxReqs: make(chan *txRequest),
quit: make(chan bool),
flushReq: make(chan flushSpendReq),
txPool: make([]MempoolTx, 0),
poolSpends: make(map[string]bool),
storage: storage,
client: client,
}
}

func (m *Mempool) validateTransactionRequest(req *txRequest) {
tx := req.tx
var err error
if tx.Body.IsDeposit() {
err = m.VerifyDepositTransaction(&tx)
} else {
// 300-500us
err = m.VerifySpendTransaction(&tx)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Once you move out the mempool.poolSpends checks out of this method and it just becomes validation.ValidateSpendTransaction (which only checks the previous block in storage and does math only, it basically can be parallelized).

I know you said the storage.isDoubleSpent can not be parallelized but exactly why? it's checking the previous persisted block only, which won't change until mempool gets flushed. And Mempool basically contains transactions for the next block and having the ensurePoolSpends separated on a single thread (under the new case req := <-m.validatedTxReqs:) ensures no double-spend happens in mempool.

In summary: storage doubleSpend check can be parallelized cause it's past history and static but maintaining mempool double-spend integrity on a single thread keeps the correctness.

Thoughts?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Second thoughts on this: realized that if a block is about to be written with transaction X (via flush), but a double-spend transaction X' is in a go-routine just past this line, it's possible for a double-spend to sneak in.

Basically mempool validation can't occur while a block is about to be written.

So you're right - I will need to re-think a little bit more

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct, mempool validation cannot occur while a block is about to be/is being written. That's handled by the flushReq case in the mempool loop - all transaction processing halts until the block is successfully written to disk.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When I said the double spend check can't be parallelized I was referring to how it requires that blocks be written to disk before new transactions can be processed by the mempool.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah but I think there is a race-condition introduced here. As described in my last comment: https://github.com/kyokan/plasma/pull/103/files#r259601343.

The transaction validation can occur simultaneously as a package block write. Basically we should wrap the go-routine in a lock.

}
if err != nil {
mPoolLogger.WithFields(logrus.Fields{
"hash": tx.Body.SignatureHash().Hex(),
"reason": err,
}).Warn("transaction rejected from mempool")

req.res <- TxInclusionResponse{
Error: err,
}
return
}

m.validatedTxReqs <- req
}

func (m *Mempool) Start() error {
go func() {
for {
select {
case req := <-m.txReqs:
if len(m.txPool) == MaxMempoolSize {
req.res <- TxInclusionResponse{
Error: errors.New("mempool is full"),
}
continue
}

tx := req.tx
var err error
if tx.Body.IsDeposit() {
err = m.VerifyDepositTransaction(&tx)
} else {
err = m.VerifySpendTransaction(&tx)
}
if err != nil {
go m.validateTransactionRequest(req)
case req := <-m.validatedTxReqs:
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We validate the incoming transaction request first against previous block history for double-spends and once that's cleared we do our mem-pool integrity checks synchronously below.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like this approach.

if len(m.txPool) == MaxMempoolSize {
req.res <- TxInclusionResponse{
Error: errors.New("mempool is full"),
}
continue
}

tx := req.tx

if err := m.ensureNoPoolSpend(&tx); err != nil {
mPoolLogger.WithFields(logrus.Fields{
"hash": tx.Body.SignatureHash().Hex(),
"reason": err,
Expand All @@ -90,10 +113,13 @@ func (m *Mempool) Start() error {
}
continue
}

m.txPool = append(m.txPool, MempoolTx{
Tx: tx,
Response: req.res,
})

// 3us
m.updatePoolSpends(&tx)
case req := <-m.flushReq:
res := m.txPool
Expand Down Expand Up @@ -134,18 +160,10 @@ func (m *Mempool) Append(tx chain.Transaction) TxInclusionResponse {
}

func (m *Mempool) VerifySpendTransaction(tx *chain.Transaction) (error) {
if err := m.ensureNoPoolSpend(tx); err != nil {
return err
}

return validation.ValidateSpendTransaction(m.storage, tx)
}

func (m *Mempool) VerifyDepositTransaction(tx *chain.Transaction) error {
if err := m.ensureNoPoolSpend(tx); err != nil {
return err
}

return validation.ValidateDepositTransaction(m.storage, m.client, tx)
}

Expand Down
11 changes: 11 additions & 0 deletions pkg/validation/validation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,31 +14,38 @@ import (
)

func ValidateSpendTransaction(storage db.Storage, tx *chain.Transaction) (error) {
// 300 ns
if tx.Body.Output0.Amount.Cmp(big.NewInt(0)) == -1 {
return NewErrNegativeOutput(0)
}
if tx.Body.Output1.Amount.Cmp(big.NewInt(0)) == -1 {
return NewErrNegativeOutput(1)
}

// 20 Us
prevTx0Conf, err := storage.FindTransactionByBlockNumTxIdx(tx.Body.Input0.BlockNumber, tx.Body.Input0.TransactionIndex)
if err == leveldb.ErrNotFound {
return NewErrTxNotFound(0, tx.Body.Input0.BlockNumber, tx.Body.Input0.TransactionIndex)
}
if err != nil {
return err
}

// 10 Us
prevTx0 := prevTx0Conf.Transaction
if prevTx0Conf.ConfirmSigs[0] != tx.Body.Input0ConfirmSig {
return NewErrConfirmSigMismatch(0)
}
sigHash0 := tx.Body.SignatureHash()

// 200 Us
prevTx0Output := prevTx0.Body.OutputAt(tx.Body.Input0.OutputIndex)
err = eth.ValidateSignature(sigHash0, tx.Sigs[0][:], prevTx0Output.Owner)
if err != nil {
return NewErrInvalidSignature(0)
}

// 450 ns
totalInput := big.NewInt(0)
totalInput = totalInput.Add(totalInput, prevTx0Output.Amount)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Another thing: if !tx.Body.Input1.IsZero() never gets tripped which will essentially could double the time because eth.valdiateSignature gets called twice.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you mean never gets tripped? You mean if Input1 is defined?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah if input1 is never defined. which is the case in our benchmark setup

Expand All @@ -60,14 +67,17 @@ func ValidateSpendTransaction(storage db.Storage, tx *chain.Transaction) (error)
}
prevTx1Output := prevTx1.Body.OutputAt(tx.Body.Input1.OutputIndex)
sigHash1 := tx.Body.SignatureHash()

err = eth.ValidateSignature(sigHash1, tx.Sigs[1][:], prevTx1Output.Owner)

if err != nil {
return NewErrInvalidSignature(1)
}

totalInput = totalInput.Add(totalInput, prevTx1Output.Amount)
}

// 600-900 ns
totalOutput := big.NewInt(0)
totalOutput = totalOutput.Add(totalOutput, tx.Body.Output0.Amount)
totalOutput = totalOutput.Add(totalOutput, tx.Body.Fee)
Expand All @@ -79,6 +89,7 @@ func ValidateSpendTransaction(storage db.Storage, tx *chain.Transaction) (error)
return NewErrInputOutputValueMismatch(totalInput, totalOutput)
}

// 20 Us
isDoubleSpent, err := storage.IsDoubleSpent(tx)
if err != nil {
return err
Expand Down