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

Jump to bottom

[storage] Single Errgroup Pool for BlockWorkers #272

Merged
merged 6 commits into from
Dec 9, 2020
Merged

[storage] Single Errgroup Pool for BlockWorkers #272

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
273ff7d
Provide errgroup to BlockWorker interface
patrick-ogrady Dec 8, 2020
bf92a76
update balance storage
patrick-ogrady Dec 8, 2020
4280a80
Change remaining block workers
patrick-ogrady Dec 8, 2020
03cc858
update mocks
patrick-ogrady Dec 9, 2020
9d117dc
first pass at fixing tests
patrick-ogrady Dec 9, 2020
65038e6
Fix tests
patrick-ogrady Dec 9, 2020
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
29 changes: 15 additions & 14 deletions mocks/storage/modules/block_worker.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

33 changes: 5 additions & 28 deletions storage/modules/balance_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,7 @@ func (b *BalanceStorage) Initialize(
// AddingBlock is called by BlockStorage when adding a block to storage.
func (b *BalanceStorage) AddingBlock(
ctx context.Context,
g *errgroup.Group,
block *types.Block,
transaction database.Transaction,
) (database.CommitWorker, error) {
Expand All @@ -194,19 +195,14 @@ func (b *BalanceStorage) AddingBlock(

// Keep track of how many new accounts have been seen so that the counter
// can be updated in a single op.
var newAccounts int
var newAccountsLock sync.Mutex

// Concurrent execution limited to runtime.NumCPU
g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
for i := range changes {
// We need to set variable before calling goroutine
// to avoid getting an updated pointer as loop iteration
// continues.
change := changes[i]
g.Go(func() error {
newAccount, err := b.UpdateBalance(
gctx,
ctx,
transaction,
change,
block.ParentBlockIdentifier,
Expand All @@ -219,25 +215,10 @@ func (b *BalanceStorage) AddingBlock(
return nil
}

newAccountsLock.Lock()
newAccounts++
newAccountsLock.Unlock()

return nil
return b.handler.AccountsSeen(ctx, transaction, 1)
})
}

if err := g.Wait(); err != nil {
return nil, err
}

// Update accounts seen
if newAccounts > 0 {
if err := b.handler.AccountsSeen(ctx, transaction, newAccounts); err != nil {
return nil, err
}
}

// Update accounts reconciled
var pending int
b.pendingReconciliationMutex.Lock(true)
Expand All @@ -259,6 +240,7 @@ func (b *BalanceStorage) AddingBlock(
// RemovingBlock is called by BlockStorage when removing a block from storage.
func (b *BalanceStorage) RemovingBlock(
ctx context.Context,
g *errgroup.Group,
block *types.Block,
transaction database.Transaction,
) (database.CommitWorker, error) {
Expand All @@ -273,15 +255,14 @@ func (b *BalanceStorage) RemovingBlock(
var staleAccountsMutex sync.Mutex

// Concurrent execution limited to runtime.NumCPU
g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
for i := range changes {
// We need to set variable before calling goroutine
// to avoid getting an updated pointer as loop iteration
// continues.
change := changes[i]
g.Go(func() error {
shouldRemove, err := b.OrphanBalance(
gctx,
ctx,
transaction,
change,
)
Expand All @@ -304,10 +285,6 @@ func (b *BalanceStorage) RemovingBlock(
})
}

if err := g.Wait(); err != nil {
return nil, err
}

return func(ctx context.Context) error {
if err := b.handler.BlockRemoved(ctx, block, changes); err != nil {
return err
Expand Down
47 changes: 31 additions & 16 deletions storage/modules/balance_storage_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"path"
"testing"

"github.com/neilotoole/errgroup"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"

Expand Down Expand Up @@ -1557,8 +1558,10 @@ func TestBlockSyncing(t *testing.T) {

t.Run("add genesis block", func(t *testing.T) {
dbTx := database.Transaction(ctx)
_, err = storage.AddingBlock(ctx, b0, dbTx)
g, gctx := errgroup.WithContext(ctx)
_, err = storage.AddingBlock(gctx, g, b0, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))

amount, err := storage.GetBalance(ctx, addr1, curr, b0.BlockIdentifier.Index)
Expand All @@ -1571,19 +1574,21 @@ func TestBlockSyncing(t *testing.T) {

t.Run("add block 1", func(t *testing.T) {
dbTx := database.Transaction(ctx)
g, gctx := errgroup.WithContext(ctx)
mockHelper.On(
"AccountBalance",
mock.Anything,
gctx,
addr1,
curr,
b0.BlockIdentifier,
).Return(
&types.Amount{Value: "1", Currency: curr},
nil,
).Once()
mockHandler.On("AccountsSeen", ctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(ctx, b1, dbTx)
mockHandler.On("AccountsSeen", gctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(gctx, g, b1, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))

amount, err := storage.GetBalance(ctx, addr1, curr, b0.BlockIdentifier.Index)
Expand All @@ -1609,20 +1614,22 @@ func TestBlockSyncing(t *testing.T) {
assert.NoError(t, err)

dbTx := database.Transaction(ctx)
g, gctx := errgroup.WithContext(ctx)
mockHelper.On(
"AccountBalance",
mock.Anything,
gctx,
addr2,
curr,
b1.BlockIdentifier,
).Return(
&types.Amount{Value: "0", Currency: curr},
nil,
).Once()
mockHandler.On("AccountsSeen", ctx, dbTx, 1).Return(nil).Once()
mockHandler.On("AccountsReconciled", ctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(ctx, b2, dbTx)
mockHandler.On("AccountsSeen", gctx, dbTx, 1).Return(nil).Once()
mockHandler.On("AccountsReconciled", gctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(gctx, g, b2, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))

amount, err := storage.GetBalance(ctx, addr1, curr, b0.BlockIdentifier.Index)
Expand Down Expand Up @@ -1659,8 +1666,10 @@ func TestBlockSyncing(t *testing.T) {

t.Run("orphan block 2", func(t *testing.T) {
dbTx := database.Transaction(ctx)
commitWorker, err := storage.RemovingBlock(ctx, b2, dbTx)
g, gctx := errgroup.WithContext(ctx)
commitWorker, err := storage.RemovingBlock(gctx, g, b2, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))
mockHandler.On("BlockRemoved", ctx, b2, mock.Anything).Return(nil).Once()
mockHandler.On("AccountsSeen", ctx, mock.Anything, -1).Return(nil).Once()
Expand Down Expand Up @@ -1694,8 +1703,10 @@ func TestBlockSyncing(t *testing.T) {

t.Run("orphan block 1", func(t *testing.T) {
dbTx := database.Transaction(ctx)
commitWorker, err := storage.RemovingBlock(ctx, b1, dbTx)
g, gctx := errgroup.WithContext(ctx)
commitWorker, err := storage.RemovingBlock(gctx, g, b1, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))
mockHandler.On("BlockRemoved", ctx, b1, mock.Anything).Return(nil).Once()
mockHandler.On("AccountsSeen", ctx, mock.Anything, -1).Return(nil).Once()
Expand Down Expand Up @@ -1724,19 +1735,21 @@ func TestBlockSyncing(t *testing.T) {

t.Run("add block 1", func(t *testing.T) {
dbTx := database.Transaction(ctx)
g, gctx := errgroup.WithContext(ctx)
mockHelper.On(
"AccountBalance",
mock.Anything,
gctx,
addr1,
curr,
b0.BlockIdentifier,
).Return(
&types.Amount{Value: "1", Currency: curr},
nil,
).Once()
mockHandler.On("AccountsSeen", ctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(ctx, b1, dbTx)
mockHandler.On("AccountsSeen", gctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(gctx, g, b1, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))

amount, err := storage.GetBalance(ctx, addr1, curr, b0.BlockIdentifier.Index)
Expand Down Expand Up @@ -1767,19 +1780,21 @@ func TestBlockSyncing(t *testing.T) {

t.Run("add block 2a", func(t *testing.T) {
dbTx := database.Transaction(ctx)
g, gctx := errgroup.WithContext(ctx)
mockHelper.On(
"AccountBalance",
mock.Anything,
gctx,
addr2,
curr,
b1.BlockIdentifier,
).Return(
&types.Amount{Value: "0", Currency: curr},
nil,
).Once()
mockHandler.On("AccountsSeen", ctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(ctx, b2a, dbTx)
mockHandler.On("AccountsSeen", gctx, dbTx, 1).Return(nil).Once()
_, err = storage.AddingBlock(gctx, g, b2a, dbTx)
assert.NoError(t, err)
assert.NoError(t, g.Wait())
assert.NoError(t, dbTx.Commit(ctx))

amount, err := storage.GetBalance(ctx, addr1, curr, b0.BlockIdentifier.Index)
Expand Down
Loading