Skip to content

Commit 73a2d09

Browse files
alexandrosfiliosalexandrosfilios
committed
Multi updates
Signed-off-by: Alexandros Filios <[email protected]>
1 parent d8c305a commit 73a2d09

File tree

12 files changed

+210
-149
lines changed

12 files changed

+210
-149
lines changed

‎platform/common/core/generic/vault/inspector.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ func (i *Inspector) SetStateMetadata(driver.Namespace, driver.PKey, driver.Metad
6565
panic("programming error: the rwset inspector is read-only")
6666
}
6767

68-
func (i *Inspector) SetStateMetadatas(ns driver.Namespace, kvs map[driver.PKey]driver.Metadata, block driver.BlockNum, txnum driver.TxNum) map[driver.PKey]error {
68+
func (i *Inspector) SetStateMetadatas(ns driver.Namespace, kvs map[driver.PKey]driver.VersionedMetadataValue) map[driver.PKey]error {
6969
panic("programming error: the rwset inspector is read-only")
7070
}
7171

‎platform/common/core/generic/vault/txidstore/cache.go

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ type cache[V driver.ValidationCode] interface {
3131
type txidStore[V driver.ValidationCode] interface {
3232
Get(txID driver.TxID) (V, string, error)
3333
Set(txID driver.TxID, code V, message string) error
34+
SetMultiple(txs []driver.ByNum[V]) error
3435
Iterator(pos interface{}) (collections.Iterator[*driver.ByNum[V]], error)
3536
}
3637

@@ -88,6 +89,17 @@ func (s *CachedStore[V]) Set(txID string, code V, message string) error {
8889
return nil
8990
}
9091

92+
func (s *CachedStore[V]) SetMultiple(txs []driver.ByNum[V]) error {
93+
s.logger.Debugf("Set values for %d txs into backed and cache", len(txs))
94+
if err := s.backed.SetMultiple(txs); err != nil {
95+
return err
96+
}
97+
for _, tx := range txs {
98+
s.cache.Add(tx.TxID, &Entry[V]{ValidationCode: tx.Code, ValidationMessage: tx.Message})
99+
}
100+
return nil
101+
}
102+
91103
func (s *CachedStore[V]) Iterator(pos interface{}) (collections.Iterator[*driver.ByNum[V]], error) {
92104
return s.backed.Iterator(pos)
93105
}

‎platform/common/core/generic/vault/vault.go

Lines changed: 85 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ type TXIDStoreReader[V driver.ValidationCode] interface {
3737
type TXIDStore[V driver.ValidationCode] interface {
3838
TXIDStoreReader[V]
3939
Set(txID driver.TxID, code V, message string) error
40+
SetMultiple(txs []driver.ByNum[V]) error
4041
Invalidate(txID driver.TxID)
4142
}
4243

@@ -289,94 +290,110 @@ func (db *Vault[V]) commitRWs(inputs ...commitInput) error {
289290
return errors.Wrapf(err, "begin update in store for txid %v failed", inputs)
290291
}
291292

292-
for _, input := range inputs {
293-
span := trace.SpanFromContext(input.ctx)
293+
if _, err := db.setStatuses(inputs, db.vcProvider.Busy()); err != nil {
294+
return err
295+
}
294296

295-
span.AddEvent("set_tx_busy")
296-
if err := db.txIDStore.Set(input.txID, db.vcProvider.Busy(), ""); err != nil {
297-
if !errors.HasCause(err, UniqueKeyViolation) {
298-
return err
297+
db.logger.Debugf("parse writes")
298+
writes := make(map[driver.Namespace]map[driver.PKey]VersionedValue)
299+
for _, input := range inputs {
300+
for ns, ws := range input.rws.Writes {
301+
vals := versionedValues(ws, input.block, input.indexInBloc)
302+
if nsWrites, ok := writes[ns]; !ok {
303+
writes[ns] = vals
304+
} else {
305+
collections.CopyMap(nsWrites, vals)
299306
}
300307
}
308+
}
301309

302-
db.logger.Debugf("parse writes [%s]", input.txID)
303-
span.AddEvent("store_writes")
304-
if discarded, err := db.storeWrites(input.ctx, input.rws.Writes, input.block, input.indexInBloc); err != nil {
305-
return errors.Wrapf(err, "failed storing writes")
306-
} else if discarded {
307-
db.logger.Infof("Discarded changes while storing writes as duplicates. Skipping...")
310+
if errs := db.storeAllWrites(writes); len(errs) == 0 {
311+
db.logger.Debugf("Successfully stored writes for %d namespaces", len(writes))
312+
} else if discarded, err := db.discard("", 0, 0, errs); err != nil {
313+
return errors.Wrapf(err, "failed storing writes")
314+
} else if discarded {
315+
db.logger.Infof("Discarded changes while storing writes as duplicates. Skipping...")
316+
for _, input := range inputs {
308317
db.txIDStore.Invalidate(input.txID)
309-
return nil
310318
}
319+
return nil
320+
}
311321

312-
db.logger.Debugf("parse meta writes [%s]", input.txID)
313-
span.AddEvent("store_meta_writes")
314-
if discarded, err := db.storeMetaWrites(input.ctx, input.rws.MetaWrites, input.block, input.indexInBloc); err != nil {
315-
return errors.Wrapf(err, "failed storing meta writes")
316-
} else if discarded {
317-
db.logger.Infof("Discarded changes while storing meta writes as duplicates. Skipping...")
322+
db.logger.Debugf("parse meta writes")
323+
metaWrites := make(map[driver.Namespace]map[driver.PKey]driver.VersionedMetadataValue)
324+
for _, input := range inputs {
325+
for ns, ws := range input.rws.MetaWrites {
326+
vals := versionedMetaValues(ws, input.block, input.indexInBloc)
327+
if nsWrites, ok := metaWrites[ns]; !ok {
328+
metaWrites[ns] = vals
329+
} else {
330+
collections.CopyMap(nsWrites, vals)
331+
}
332+
}
333+
}
334+
if errs := db.storeAllMetaWrites(metaWrites); len(errs) == 0 {
335+
db.logger.Debugf("Successfully stored meta writes for %d namespaces", len(metaWrites))
336+
} else if discarded, err := db.discard("", 0, 0, errs); err != nil {
337+
return errors.Wrapf(err, "failed storing meta writes")
338+
} else if discarded {
339+
db.logger.Infof("Discarded changes while storing meta writes as duplicates. Skipping...")
340+
for _, input := range inputs {
318341
db.txIDStore.Invalidate(input.txID)
319-
return nil
320342
}
343+
return nil
344+
}
321345

322-
db.logger.Debugf("set state to valid [%s]", input.txID)
323-
span.AddEvent("set_tx_valid")
324-
if discarded, err := db.setTxValid(input.txID); err != nil {
325-
return errors.Wrapf(err, "failed setting tx state to valid")
326-
} else if discarded {
327-
db.logger.Infof("Discarded changes while setting tx state to valid as duplicates. Skipping...")
328-
return nil
346+
if discarded, err := db.setStatuses(inputs, db.vcProvider.Valid()); err != nil {
347+
if err1 := db.store.Discard(); err1 != nil {
348+
db.logger.Errorf("got error %s; discarding caused %s", err.Error(), err1.Error())
329349
}
330-
350+
for _, input := range inputs {
351+
db.txIDStore.Invalidate(input.txID)
352+
}
353+
return errors.Wrapf(err, "failed setting tx state to valid")
354+
} else if discarded {
355+
if err1 := db.store.Discard(); err1 != nil {
356+
db.logger.Errorf("got unique key violation; discarding caused %s", err1.Error())
357+
}
358+
db.logger.Infof("Discarded changes while setting tx state to valid as duplicates. Skipping...")
359+
return nil
331360
}
332361

333-
for _, input := range inputs {
334-
trace.SpanFromContext(input.ctx).AddEvent("commit_update")
335-
}
336362
if err := db.store.Commit(); err != nil {
337363
return errors.Wrapf(err, "committing tx for txid in store [%v] failed", inputs)
338364
}
339-
340365
return nil
341366
}
342367

343-
func (db *Vault[V]) setTxValid(txID driver.TxID) (bool, error) {
344-
err := db.txIDStore.Set(txID, db.vcProvider.Valid(), "")
345-
if err == nil {
346-
return false, nil
368+
func (db *Vault[V]) setStatuses(inputs []commitInput, v V) (bool, error) {
369+
txs := make([]driver.ByNum[V], len(inputs))
370+
for i, input := range inputs {
371+
txs[i] = driver.ByNum[V]{TxID: input.txID, Code: v}
347372
}
348373

349-
if err1 := db.store.Discard(); err1 != nil {
350-
db.logger.Errorf("got error %s; discarding caused %s", err.Error(), err1.Error())
351-
}
352-
353-
if !errors.HasCause(err, UniqueKeyViolation) {
354-
db.txIDStore.Invalidate(txID)
355-
return true, errors.Wrapf(err, "error setting tx valid")
374+
if err := db.txIDStore.SetMultiple(txs); err == nil {
375+
return false, nil
376+
} else if !errors.HasCause(err, UniqueKeyViolation) {
377+
return true, err
378+
} else {
379+
return true, nil
356380
}
357-
return true, nil
358381
}
359382

360-
func (db *Vault[V]) storeMetaWrites(ctx context.Context, writes NamespaceKeyedMetaWrites, block driver.BlockNum, indexInBloc driver.TxNum) (bool, error) {
361-
span := trace.SpanFromContext(ctx)
362-
for ns, keyMap := range writes {
363-
span.AddEvent("set_tx_metadata_state")
364-
if errs := db.store.SetStateMetadatas(ns, keyMap, block, indexInBloc); len(errs) > 0 {
365-
return db.discard(ns, block, indexInBloc, errs)
366-
}
383+
func (db *Vault[V]) storeAllWrites(writes map[driver.Namespace]map[driver.PKey]VersionedValue) map[driver.PKey]error {
384+
errs := make(map[driver.PKey]error)
385+
for ns, vals := range writes {
386+
collections.CopyMap(errs, db.store.SetStates(ns, vals))
367387
}
368-
return false, nil
388+
return errs
369389
}
370390

371-
func (db *Vault[V]) storeWrites(ctx context.Context, writes Writes, block driver.BlockNum, indexInBloc driver.TxNum) (bool, error) {
372-
span := trace.SpanFromContext(ctx)
373-
for ns, keyMap := range writes {
374-
span.AddEvent("set_tx_states")
375-
if errs := db.store.SetStates(ns, versionedValues(keyMap, block, indexInBloc)); len(errs) > 0 {
376-
return db.discard(ns, block, indexInBloc, errs)
377-
}
391+
func (db *Vault[V]) storeAllMetaWrites(metaWrites map[driver.Namespace]map[driver.PKey]driver.VersionedMetadataValue) map[driver.PKey]error {
392+
errs := make(map[driver.PKey]error)
393+
for ns, vals := range metaWrites {
394+
collections.CopyMap(errs, db.store.SetStateMetadatas(ns, vals))
378395
}
379-
return false, nil
396+
return errs
380397
}
381398

382399
func versionedValues(keyMap NamespaceWrites, block driver.BlockNum, indexInBloc driver.TxNum) map[driver.PKey]VersionedValue {
@@ -387,6 +404,14 @@ func versionedValues(keyMap NamespaceWrites, block driver.BlockNum, indexInBloc
387404
return vals
388405
}
389406

407+
func versionedMetaValues(keyMap KeyedMetaWrites, block driver.BlockNum, indexInBloc driver.TxNum) map[driver.PKey]driver.VersionedMetadataValue {
408+
vals := make(map[driver.PKey]driver.VersionedMetadataValue, len(keyMap))
409+
for pkey, val := range keyMap {
410+
vals[pkey] = driver.VersionedMetadataValue{Metadata: val, Block: block, TxNum: indexInBloc}
411+
}
412+
return vals
413+
}
414+
390415
func (db *Vault[V]) discard(ns driver.Namespace, block driver.BlockNum, indexInBloc driver.TxNum, errs map[driver.PKey]error) (bool, error) {
391416
if err1 := db.store.Discard(); err1 != nil {
392417
db.logger.Errorf("got error %v; discarding caused %s", errors2.Join(collections.Values(errs)...), err1.Error())

‎platform/common/driver/vault.go

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,12 @@ type VersionedRead struct {
2626
TxNum TxNum
2727
}
2828

29+
type VersionedMetadataValue struct {
30+
Block BlockNum
31+
TxNum TxNum
32+
Metadata Metadata
33+
}
34+
2935
type VersionedResultsIterator = collections.Iterator[*VersionedRead]
3036

3137
type QueryExecutor interface {

‎platform/fabric/driver/iterator.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,4 +36,5 @@ type TXIDStore interface {
3636
Iterator(pos interface{}) (TxIDIterator, error)
3737
Get(txid string) (ValidationCode, string, error)
3838
Set(txID string, code ValidationCode, message string) error
39+
SetMultiple(txs []driver.ByNum[ValidationCode]) error
3940
}

‎platform/view/services/db/driver/badger/badger.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -173,10 +173,10 @@ func (db *DB) SetStateMetadata(namespace driver2.Namespace, key driver2.PKey, me
173173
return nil
174174
}
175175

176-
func (db *DB) SetStateMetadatas(ns driver2.Namespace, kvs map[driver2.PKey]driver2.Metadata, block driver2.BlockNum, txnum driver2.TxNum) map[driver2.PKey]error {
176+
func (db *DB) SetStateMetadatas(ns driver2.Namespace, kvs map[driver2.PKey]driver2.VersionedMetadataValue) map[driver2.PKey]error {
177177
errs := make(map[driver2.PKey]error)
178178
for pkey, value := range kvs {
179-
if err := db.SetStateMetadata(ns, pkey, value, block, txnum); err != nil {
179+
if err := db.SetStateMetadata(ns, pkey, value.Metadata, value.Block, value.TxNum); err != nil {
180180
errs[pkey] = err
181181
}
182182
}

‎platform/view/services/db/driver/driver.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@ type VersionedValue struct {
2727
TxNum driver.TxNum
2828
}
2929

30+
type VersionedMetadataValue = driver.VersionedMetadataValue
31+
3032
type UnversionedRead struct {
3133
Key driver.PKey
3234
Raw driver.RawValue
@@ -90,7 +92,7 @@ type VersionedPersistence interface {
9092
// SetStateMetadata sets the given metadata for the given namespace, key, and version
9193
SetStateMetadata(namespace driver.Namespace, key driver.PKey, metadata driver.Metadata, block driver.BlockNum, txnum driver.TxNum) error
9294
// SetStateMetadatas sets the given metadata for the given namespace, keys, and version
93-
SetStateMetadatas(ns driver.Namespace, kvs map[driver.PKey]driver.Metadata, block driver.BlockNum, txnum driver.TxNum) map[driver.PKey]error
95+
SetStateMetadatas(ns driver.Namespace, kvs map[driver.PKey]driver.VersionedMetadataValue) map[driver.PKey]error
9496
}
9597

9698
type WriteTransaction interface {

‎platform/view/services/db/driver/notifier/persistence.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -199,8 +199,8 @@ func (db *VersionedPersistenceNotifier[P]) SetStateMetadata(namespace driver2.Na
199199
return db.Persistence.SetStateMetadata(namespace, key, metadata, block, txnum)
200200
}
201201

202-
func (db *VersionedPersistenceNotifier[P]) SetStateMetadatas(ns driver2.Namespace, kvs map[driver2.PKey]driver2.Metadata, block driver2.BlockNum, txnum driver2.TxNum) map[driver2.PKey]error {
203-
return db.Persistence.SetStateMetadatas(ns, kvs, block, txnum)
202+
func (db *VersionedPersistenceNotifier[P]) SetStateMetadatas(ns driver2.Namespace, kvs map[driver2.PKey]driver2.VersionedMetadataValue) map[driver2.PKey]error {
203+
return db.Persistence.SetStateMetadatas(ns, kvs)
204204
}
205205

206206
func (db *VersionedPersistenceNotifier[P]) GetStateRangeScanIterator(namespace driver2.Namespace, startKey, endKey driver2.PKey) (driver.VersionedResultsIterator, error) {

‎platform/view/services/db/driver/sql/common/base.go

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ type readScanner[V any] interface {
3434
ReadValue(scannable) (V, error)
3535
}
3636

37-
type valueScanner[V any] interface {
37+
type ValueScanner[V any] interface {
3838
readScanner[V]
3939
// WriteValue writes the values of the V struct in the order given by the Columns method
4040
WriteValue(V) []any
@@ -47,12 +47,12 @@ type BasePersistence[V any, R any] struct {
4747
table string
4848

4949
readScanner readScanner[R]
50-
ValueScanner valueScanner[V]
50+
ValueScanner ValueScanner[V]
5151
errorWrapper driver.SQLErrorWrapper
5252
ci Interpreter
5353
}
5454

55-
func NewBasePersistence[V any, R any](writeDB *sql.DB, readDB *sql.DB, table string, readScanner readScanner[R], valueScanner valueScanner[V], errorWrapper driver.SQLErrorWrapper, ci Interpreter, newTransaction func() (*sql.Tx, error)) *BasePersistence[V, R] {
55+
func NewBasePersistence[V any, R any](writeDB *sql.DB, readDB *sql.DB, table string, readScanner readScanner[R], valueScanner ValueScanner[V], errorWrapper driver.SQLErrorWrapper, ci Interpreter, newTransaction func() (*sql.Tx, error)) *BasePersistence[V, R] {
5656
return &BasePersistence[V, R]{
5757
BaseDB: common.NewBaseDB[*sql.Tx](func() (*sql.Tx, error) { return newTransaction() }),
5858
readDB: readDB,
@@ -207,6 +207,20 @@ func (db *BasePersistence[V, R]) SetStateWithTx(tx *sql.Tx, ns driver2.Namespace
207207
val = append([]byte(nil), val...)
208208
values[valIndex] = val
209209

210+
return db.UpsertStateWithTx(tx, ns, pkey, keys, values)
211+
}
212+
213+
func (db *BasePersistence[V, R]) UpsertStates(ns driver2.Namespace, valueKeys []string, vals map[driver2.PKey][]any) map[driver2.PKey]error {
214+
errs := make(map[driver2.PKey]error)
215+
for pkey, val := range vals {
216+
if err := db.UpsertStateWithTx(db.Txn, ns, pkey, valueKeys, val); err != nil {
217+
errs[pkey] = err
218+
}
219+
}
220+
return errs
221+
}
222+
223+
func (db *BasePersistence[V, R]) UpsertStateWithTx(tx *sql.Tx, ns driver2.Namespace, pkey driver2.PKey, keys []string, values []any) error {
210224
// Portable upsert
211225
exists, err := db.exists(tx, ns, pkey)
212226
if err != nil {

0 commit comments

Comments
 (0)