From 12abd57c49fc4e7136ef8785608f36dd0a7154ee Mon Sep 17 00:00:00 2001 From: Liran Funaro Date: Sun, 29 Jan 2023 10:55:19 +0200 Subject: [PATCH 1/2] Perform mvcc read validation in parallel Signed-off-by: Liran Funaro --- internal/txvalidation/data_tx_validator.go | 139 ++++++++++++++++++++- internal/txvalidation/validator.go | 24 +++- internal/txvalidation/validator_test.go | 5 +- 3 files changed, 163 insertions(+), 5 deletions(-) diff --git a/internal/txvalidation/data_tx_validator.go b/internal/txvalidation/data_tx_validator.go index b4f70262..b30c74fe 100644 --- a/internal/txvalidation/data_tx_validator.go +++ b/internal/txvalidation/data_tx_validator.go @@ -113,7 +113,7 @@ func (v *dataTxValidator) parallelValidation(txsEnv []*types.DataTxEnvelope, use } func (v *dataTxValidator) validateSignatures(txEnv *types.DataTxEnvelope) ([]string, *types.ValidationInfo, error) { - var userIDsWithValidSign []string + userIDsWithValidSign := make([]string, 0, len(txEnv.Signatures)) for userID, signature := range txEnv.Signatures { valRes, err := v.sigValidator.validate(userID, signature, txEnv.Payload) if err != nil { @@ -484,6 +484,108 @@ func (v *dataTxValidator) validateACLForWriteOrDelete(userIDs []string, dbName, }, nil } +type readCache struct { + dbName string + key string + ver *types.Version + err error + wg sync.WaitGroup +} + +func (v *dataTxValidator) parallelReadMvccValidation( + valInfoArray []*types.ValidationInfo, dataTxEnvs []*types.DataTxEnvelope, +) error { + reads := make(map[string]map[string]*readCache) + errorChan := make(chan error) + + // Submit a "get-version" Go routine for each key in the envelope. + // We avoid reading the same key twice. + for txNum, txEnv := range dataTxEnvs { + if valInfoArray[txNum].Flag != types.Flag_VALID { + continue + } + for _, txOps := range txEnv.Payload.DbOperations { + dbName := txOps.DbName + dbReads, ok := reads[dbName] + if !ok { + dbReads = make(map[string]*readCache) + reads[dbName] = dbReads + } + + for _, r := range txOps.DataReads { + key := r.Key + if _, ok := dbReads[key]; ok { + continue + } + + c := &readCache{ + dbName: dbName, + key: key, + } + c.wg.Add(1) + dbReads[key] = c + go func(txNum int, c *readCache) { + defer c.wg.Done() + c.ver, c.err = v.db.GetVersion(c.dbName, c.key) + if c.err != nil { + v.logger.Errorf("error validating signatures in tx number %d, error: %s", txNum, c.err) + defer func() { + // Ignore panic when errorChan is closed + recover() + }() + errorChan <- c.err + } + }(txNum, c) + } + } + } + + // Submit a "validation" Go routine for read operation in the envelope. + wg := sync.WaitGroup{} + for txNum, txEnv := range dataTxEnvs { + for _, txOps := range txEnv.Payload.DbOperations { + for _, r := range txOps.DataReads { + if valInfoArray[txNum].Flag != types.Flag_VALID { + continue + } + + wg.Add(1) + go func(txNum int, c *readCache, expectedVer *types.Version) { + defer wg.Done() + if c == nil { + panic("all reads keys should be in the map") + } + + c.wg.Wait() + if valInfoArray[txNum].Flag != types.Flag_VALID || c.err != nil { + return + } + if proto.Equal(expectedVer, c.ver) { + return + } + valInfoArray[txNum] = &types.ValidationInfo{ + Flag: types.Flag_INVALID_MVCC_CONFLICT_WITH_COMMITTED_STATE, + ReasonIfInvalid: "mvcc conflict has occurred as the committed state for the key [" + c.key + "] in database [" + c.dbName + "] changed", + } + }(txNum, reads[txOps.DbName][r.Key], r.Version) + } + } + } + + // Wait for all the validation routines to end. + go func() { + wg.Wait() + // Inject nil to make sure we have data to read from the channel if no error occurred. + errorChan <- nil + }() + + select { + case err := <-errorChan: + close(errorChan) + return err + } +} + func (v *dataTxValidator) mvccValidation(dbName string, txOps *types.DBOperation, pendingOps *pendingOperations) (*types.ValidationInfo, error) { for _, r := range txOps.DataReads { if pendingOps.exist(dbName, r.Key) { @@ -532,3 +634,38 @@ func (v *dataTxValidator) mvccValidation(dbName string, txOps *types.DBOperation Flag: types.Flag_VALID, }, nil } + +func (v *dataTxValidator) mvccValidationPending(dbName string, txOps *types.DBOperation, pendingOps *pendingOperations) (*types.ValidationInfo, error) { + for _, r := range txOps.DataReads { + if pendingOps.exist(dbName, r.Key) { + return &types.ValidationInfo{ + Flag: types.Flag_INVALID_MVCC_CONFLICT_WITHIN_BLOCK, + ReasonIfInvalid: "mvcc conflict has occurred within the block for the key [" + r.Key + "] in database [" + dbName + "]", + }, nil + } + } + // as state trie generation work at the boundary of block, we cannot allow more than one write per key. This is because, the state trie + // generation considers only the final updates and not intermediate updates within a block boundary. As a result, we would have intermediate + // entries in the provenance store but cannot generate proof of existence for the same using the state trie. As blind writes/deletes are quite + // rare, we allow only one write per key within a block. In general, user reads the key before writing to it. + for _, w := range txOps.DataWrites { + if pendingOps.exist(dbName, w.Key) { + return &types.ValidationInfo{ + Flag: types.Flag_INVALID_MVCC_CONFLICT_WITHIN_BLOCK, + ReasonIfInvalid: "mvcc conflict has occurred within the block for the key [" + w.Key + "] in database [" + dbName + "]. Within a block, a key can be modified only once", + }, nil + } + } + for _, d := range txOps.DataDeletes { + if pendingOps.exist(dbName, d.Key) { + return &types.ValidationInfo{ + Flag: types.Flag_INVALID_MVCC_CONFLICT_WITHIN_BLOCK, + ReasonIfInvalid: "mvcc conflict has occurred within the block for the key [" + d.Key + "] in database [" + dbName + "]. Within a block, a key can be modified only once", + }, nil + } + } + + return &types.ValidationInfo{ + Flag: types.Flag_VALID, + }, nil +} diff --git a/internal/txvalidation/validator.go b/internal/txvalidation/validator.go index 00b21ac2..a7532feb 100644 --- a/internal/txvalidation/validator.go +++ b/internal/txvalidation/validator.go @@ -91,8 +91,26 @@ func (v *Validator) ValidateBlock(block *types.Block) ([]*types.ValidationInfo, return nil, err } - if err = v.dataTxValidator.parallelValidation(dataTxEnvs, usersWithValidSigPerTX, valInfoArray); err != nil { - return nil, errors.WithMessage(err, "error while validating data transaction") + var wg sync.WaitGroup + var parallelValidErr error + var parallelMvccReadErr error + wg.Add(2) + go func() { + defer wg.Done() + parallelValidErr = v.dataTxValidator.parallelValidation(dataTxEnvs, usersWithValidSigPerTX, valInfoArray) + }() + go func() { + defer wg.Done() + parallelMvccReadErr = v.dataTxValidator.parallelReadMvccValidation(valInfoArray, dataTxEnvs) + }() + wg.Wait() + + if parallelValidErr != nil { + return nil, errors.WithMessage(parallelValidErr, "error while validating data transaction") + } + + if parallelMvccReadErr != nil { + return nil, errors.WithMessage(parallelMvccReadErr, "error while validating data transaction's read set") } pendingOps := newPendingOperations() @@ -116,7 +134,7 @@ func (v *Validator) ValidateBlock(block *types.Block) ([]*types.ValidationInfo, go func() { defer wg.Done() - mvccValResult, mvccValError = v.dataTxValidator.mvccValidation(txOps.DbName, txOps, pendingOps) + mvccValResult, mvccValError = v.dataTxValidator.mvccValidationPending(txOps.DbName, txOps, pendingOps) }() wg.Wait() diff --git a/internal/txvalidation/validator_test.go b/internal/txvalidation/validator_test.go index 2d32cf77..4054e9e1 100644 --- a/internal/txvalidation/validator_test.go +++ b/internal/txvalidation/validator_test.go @@ -15,6 +15,7 @@ import ( "github.com/hyperledger-labs/orion-server/pkg/logger" "github.com/hyperledger-labs/orion-server/pkg/server/testutils" "github.com/hyperledger-labs/orion-server/pkg/types" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -657,7 +658,9 @@ func TestValidateDataBlock(t *testing.T) { results, err := env.validator.ValidateBlock(tt.block) require.NoError(t, err) - require.Equal(t, tt.expectedResults, results) + for i := range tt.expectedResults { + assert.Equal(t, tt.expectedResults[i], results[i], "index: %d", i) + } }) } } From 424b7fce1aa4f47117467fc14aa1de480c2ca529 Mon Sep 17 00:00:00 2001 From: Liran Funaro Date: Mon, 6 Feb 2023 12:04:30 +0200 Subject: [PATCH 2/2] Mostly readability modifications Signed-off-by: Liran Funaro --- internal/txvalidation/data_tx_validator.go | 62 +++++++++++++--------- 1 file changed, 38 insertions(+), 24 deletions(-) diff --git a/internal/txvalidation/data_tx_validator.go b/internal/txvalidation/data_tx_validator.go index b30c74fe..0a03b168 100644 --- a/internal/txvalidation/data_tx_validator.go +++ b/internal/txvalidation/data_tx_validator.go @@ -497,14 +497,19 @@ func (v *dataTxValidator) parallelReadMvccValidation( ) error { reads := make(map[string]map[string]*readCache) errorChan := make(chan error) + defer close(errorChan) - // Submit a "get-version" Go routine for each key in the envelope. + // Submit a "get-version" Go routine for each read key in the envelope. // We avoid reading the same key twice. for txNum, txEnv := range dataTxEnvs { if valInfoArray[txNum].Flag != types.Flag_VALID { continue } for _, txOps := range txEnv.Payload.DbOperations { + if len(txOps.DataReads) == 0 { + continue + } + dbName := txOps.DbName dbReads, ok := reads[dbName] if !ok { @@ -513,25 +518,23 @@ func (v *dataTxValidator) parallelReadMvccValidation( } for _, r := range txOps.DataReads { - key := r.Key - if _, ok := dbReads[key]; ok { + if _, ok := dbReads[r.Key]; ok { continue } c := &readCache{ dbName: dbName, - key: key, + key: r.Key, } c.wg.Add(1) - dbReads[key] = c + dbReads[r.Key] = c go func(txNum int, c *readCache) { defer c.wg.Done() c.ver, c.err = v.db.GetVersion(c.dbName, c.key) if c.err != nil { v.logger.Errorf("error validating signatures in tx number %d, error: %s", txNum, c.err) defer func() { - // Ignore panic when errorChan is closed - recover() + recover() // Ignore panic when errorChan is closed }() errorChan <- c.err } @@ -543,45 +546,56 @@ func (v *dataTxValidator) parallelReadMvccValidation( // Submit a "validation" Go routine for read operation in the envelope. wg := sync.WaitGroup{} for txNum, txEnv := range dataTxEnvs { + if valInfoArray[txNum].Flag != types.Flag_VALID { + continue + } for _, txOps := range txEnv.Payload.DbOperations { + if len(txOps.DataReads) == 0 { + continue + } + dbReads, ok := reads[txOps.DbName] + if !ok { + panic("all read DBs should be in the map") + } for _, r := range txOps.DataReads { - if valInfoArray[txNum].Flag != types.Flag_VALID { - continue + keyRead, ok := dbReads[r.Key] + if !ok { + panic("all read keys should be in the map") } wg.Add(1) go func(txNum int, c *readCache, expectedVer *types.Version) { defer wg.Done() - if c == nil { - panic("all reads keys should be in the map") + // Stop early in case another validation routine already invalidated this TX. + if valInfoArray[txNum].Flag != types.Flag_VALID { + return } c.wg.Wait() - if valInfoArray[txNum].Flag != types.Flag_VALID || c.err != nil { - return - } - if proto.Equal(expectedVer, c.ver) { - return - } - valInfoArray[txNum] = &types.ValidationInfo{ - Flag: types.Flag_INVALID_MVCC_CONFLICT_WITH_COMMITTED_STATE, - ReasonIfInvalid: "mvcc conflict has occurred as the committed state for the key [" + c.key + "] in database [" + c.dbName + "] changed", + + // We check the flag again after waiting for the read version. + // The version comparison is last to avoid redundant comparison (short circuit evaluation). + if c.err == nil && valInfoArray[txNum].Flag == types.Flag_VALID && !proto.Equal(expectedVer, c.ver) { + valInfoArray[txNum] = &types.ValidationInfo{ + Flag: types.Flag_INVALID_MVCC_CONFLICT_WITH_COMMITTED_STATE, + ReasonIfInvalid: "mvcc conflict has occurred as the committed state for the key [" + c.key + "] in database [" + c.dbName + "] changed", + } } - }(txNum, reads[txOps.DbName][r.Key], r.Version) + }(txNum, keyRead, r.Version) } } } - // Wait for all the validation routines to end. + // Wait in the background for all the validation routines to end, then inject nil to make sure we have data + // to read from the channel if no error occurred. go func() { wg.Wait() - // Inject nil to make sure we have data to read from the channel if no error occurred. errorChan <- nil }() + // Wait for all the validation routines to end or for the first error. select { case err := <-errorChan: - close(errorChan) return err } }