From abe668ec00cb8c872b7ced50bfda05b0c2d184f4 Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Thu, 12 Nov 2020 18:57:27 -0500 Subject: [PATCH 01/10] refactor merkle trie rebuild --- ledger/accountdb.go | 183 ++++++++++++++++++++++++++++++++++++++++++ ledger/acctupdates.go | 87 +++++++++++--------- 2 files changed, 231 insertions(+), 39 deletions(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index 8ea1c4b626..0f658802e9 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1131,3 +1131,186 @@ func (iterator *encodedAccountsBatchIter) Close() { iterator.rows = nil } } + +// orderedAccountsBuilderIter allows us to iterate over the accounts addresses in the order of the account hashes. +type orderedAccountsBuilderIter struct { + step int + rows *sql.Rows + tx *sql.Tx + accountCount int + accountData bool + insertStmt *sql.Stmt +} + +func makeOrderedAccountsBuilderIter(tx *sql.Tx, accountCount int, accountData bool) *orderedAccountsBuilderIter { + return &orderedAccountsBuilderIter{ + tx: tx, + accountCount: accountCount, + accountData: accountData, + step: 0, + } +} + +type accountAddressHashData struct { + address basics.Address + digest []byte + encodedAccountData []byte +} + +func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []accountAddressHashData, ordered bool, err error) { + if iterator.step == 0 { + _, err = iterator.tx.ExecContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") + if err != nil { + return + } + iterator.step = 1 + return + } + if iterator.step == 1 { + _, err = iterator.tx.ExecContext(ctx, "CREATE TABLE accountsiteratorhashes(address blob, hash blob)") + if err != nil { + return + } + iterator.step = 2 + return + } + if iterator.step == 2 { + iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT address, data FROM accountbase") + if err != nil { + return + } + iterator.insertStmt, err = iterator.tx.PrepareContext(ctx, "INSERT INTO accountsiteratorhashes(address, hash) VALUES(?, ?)") + if err != nil { + return + } + iterator.step = 3 + return + } + if iterator.step == 3 { + var addr basics.Address + count := 0 + for iterator.rows.Next() { + var addrbuf []byte + var buf []byte + err = iterator.rows.Scan(&addrbuf, &buf) + if err != nil { + iterator.Close(ctx) + return + } + + if len(addrbuf) != len(addr) { + err = fmt.Errorf("Account DB address length mismatch: %d != %d", len(addrbuf), len(addr)) + return + } + + copy(addr[:], addrbuf) + + var accountData basics.AccountData + err = protocol.Decode(buf, &accountData) + if err != nil { + return + } + hash := accountHashBuilder(addr, accountData, buf) + _, err = iterator.insertStmt.ExecContext(ctx, addrbuf, hash) + if err != nil { + return + } + + count++ + if count == iterator.accountCount { + // we're done with this iteration. + acct = make([]accountAddressHashData, count, count) + return + } + } + acct = make([]accountAddressHashData, count, count) + iterator.step = 4 + iterator.rows.Close() + iterator.rows = nil + iterator.insertStmt.Close() + iterator.insertStmt = nil + return + } + if iterator.step == 4 { + _, err = iterator.tx.ExecContext(ctx, "CREATE INDEX accountsiteratorhashesidx ON accountsiteratorhashes(hash)") + if err != nil { + return + } + iterator.step = 5 + return + } + if iterator.step == 5 { + if iterator.accountData { + iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT accountsiteratorhashes.address, accountsiteratorhashes.hash, accountbase.data FROM accountsiteratorhashes JOIN accountbase ON accountbase.address=accountsiteratorhashes.address ORDER BY accountsiteratorhashes.hash") + } else { + iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT address, hash FROM accountsiteratorhashes ORDER BY hash") + } + + if err != nil { + return + } + iterator.step = 6 + return + } + + if iterator.step == 6 { + acct = make([]accountAddressHashData, 0, iterator.accountCount) + var addr basics.Address + for iterator.rows.Next() { + var addrbuf []byte + var acctdata []byte + var hash []byte + if iterator.accountData { + err = iterator.rows.Scan(&addrbuf, &hash, &acctdata) + } else { + err = iterator.rows.Scan(&addrbuf, &hash) + } + if err != nil { + iterator.Close(ctx) + return + } + + if len(addrbuf) != len(addr) { + err = fmt.Errorf("Account DB address length mismatch: %d != %d", len(addrbuf), len(addr)) + return + } + + copy(addr[:], addrbuf) + + acct = append(acct, accountAddressHashData{address: addr, digest: hash, encodedAccountData: acctdata}) + if len(acct) == iterator.accountCount { + // we're done with this iteration. + ordered = true + return + } + } + ordered = true + iterator.step = 7 + iterator.rows.Close() + iterator.rows = nil + return + } + if iterator.step == 7 { + iterator.rows, err = iterator.tx.QueryContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") + if err != nil { + return + } + iterator.step = 8 + // fallthrough + } + return nil, true, sql.ErrNoRows +} + +// Close shuts down the orderedAccountsBuilderIter, releasing database resources. +func (iterator *orderedAccountsBuilderIter) Close(ctx context.Context) (err error) { + if iterator.rows != nil { + iterator.rows.Close() + iterator.rows = nil + } + if iterator.insertStmt != nil { + iterator.insertStmt.Close() + iterator.insertStmt = nil + } + _, err = iterator.tx.ExecContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") + return +} diff --git a/ledger/acctupdates.go b/ledger/acctupdates.go index 21519e4200..3eb4849701 100644 --- a/ledger/acctupdates.go +++ b/ledger/acctupdates.go @@ -1119,56 +1119,65 @@ func (au *accountUpdates) accountsInitialize(ctx context.Context, tx *sql.Tx) (b if rootHash.IsZero() { au.log.Infof("accountsInitialize rebuilding merkle trie for round %d", rnd) - var accountsIterator encodedAccountsBatchIter - defer accountsIterator.Close() + accountBuilderIt := makeOrderedAccountsBuilderIter(tx, trieRebuildAccountChunkSize, false) + defer accountBuilderIt.Close(ctx) startTrieBuildTime := time.Now() accountsCount := 0 lastRebuildTime := startTrieBuildTime pendingAccounts := 0 + wasOrdered := false for { - bal, err := accountsIterator.Next(ctx, tx, trieRebuildAccountChunkSize) - if err != nil { - return rnd, err - } - if len(bal) == 0 { + accts, ordered, err := accountBuilderIt.Next(ctx) + if err == sql.ErrNoRows { + // the account builder would return sql.ErrNoRows when no more data is available. break - } - accountsCount += len(bal) - pendingAccounts += len(bal) - for _, balance := range bal { - var accountData basics.AccountData - err = protocol.Decode(balance.AccountData, &accountData) - if err != nil { - return rnd, err - } - hash := accountHashBuilder(balance.Address, accountData, balance.AccountData) - added, err := trie.Add(hash) - if err != nil { - return rnd, fmt.Errorf("accountsInitialize was unable to add changes to trie: %v", err) - } - if !added { - au.log.Warnf("accountsInitialize attempted to add duplicate hash '%s' to merkle trie for account %v", hex.EncodeToString(hash), balance.Address) - } + } else if err != nil { + return rnd, err } - if pendingAccounts >= trieRebuildCommitFrequency { - // this trie Evict will commit using the current transaction. - // if anything goes wrong, it will still get rolled back. - _, err = trie.Evict(true) - if err != nil { - return 0, fmt.Errorf("accountsInitialize was unable to commit changes to trie: %v", err) - } - pendingAccounts = 0 + if !wasOrdered && ordered { + accountsCount = 0 + wasOrdered = true } - if len(bal) < trieRebuildAccountChunkSize { - break - } + accountsCount += len(accts) + if len(accts) > 0 { + if ordered { + pendingAccounts += len(accts) + for _, acct := range accts { + added, err := trie.Add(acct.digest) + if err != nil { + return rnd, fmt.Errorf("accountsInitialize was unable to add changes to trie: %v", err) + } + if !added { + au.log.Warnf("accountsInitialize attempted to add duplicate hash '%s' to merkle trie for account %v", hex.EncodeToString(acct.digest), acct.address) + } + } + + if pendingAccounts >= trieRebuildCommitFrequency { + // this trie Evict will commit using the current transaction. + // if anything goes wrong, it will still get rolled back. + _, err = trie.Evict(true) + if err != nil { + return 0, fmt.Errorf("accountsInitialize was unable to commit changes to trie: %v", err) + } + pendingAccounts = 0 + } - if time.Now().Sub(lastRebuildTime) > 5*time.Second { - // let the user know that the trie is still being rebuilt. - au.log.Infof("accountsInitialize still building the trie, and processed so far %d accounts", accountsCount) - lastRebuildTime = time.Now() + if time.Now().Sub(lastRebuildTime) > 5*time.Second { + // let the user know that the trie is still being rebuilt. + au.log.Infof("accountsInitialize still building the trie, and processed so far %d accounts", accountsCount) + lastRebuildTime = time.Now() + } + + } else { + // if it's not ordered, we can ignore it for now; we'll just increase the counters and emit logs periodically. + if time.Now().Sub(lastRebuildTime) > 5*time.Second { + // let the user know that the trie is still being rebuilt. + au.log.Infof("accountsInitialize still building the trie, and hashed so far %d accounts", accountsCount) + lastRebuildTime = time.Now() + } + } } } From a67e49076fd9a2c5e8ea1ab18c0e67ad17afea45 Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 10:32:40 -0500 Subject: [PATCH 02/10] Small refactor --- ledger/accountdb.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index 0f658802e9..c25e71f330 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1224,11 +1224,11 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac } } acct = make([]accountAddressHashData, count, count) - iterator.step = 4 iterator.rows.Close() iterator.rows = nil iterator.insertStmt.Close() iterator.insertStmt = nil + iterator.step = 4 return } if iterator.step == 4 { @@ -1291,7 +1291,7 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac return } if iterator.step == 7 { - iterator.rows, err = iterator.tx.QueryContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") + err = iterator.Close(ctx) if err != nil { return } From aaa31e2a12bd29c4469795a3eec1daa4ec801eae Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 10:33:07 -0500 Subject: [PATCH 03/10] fix grammer --- ledger/catchpointwriter_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ledger/catchpointwriter_test.go b/ledger/catchpointwriter_test.go index d39f3a1615..37fd63f98e 100644 --- a/ledger/catchpointwriter_test.go +++ b/ledger/catchpointwriter_test.go @@ -124,7 +124,7 @@ func TestCatchpointFileBalancesChunkEncoding(t *testing.T) { } func TestBasicCatchpointWriter(t *testing.T) { - // create new protocol version, which has lower back balance. + // create new protocol version, which has lower lookback testProtocolVersion := protocol.ConsensusVersion("test-protocol-TestBasicCatchpointWriter") protoParams := config.Consensus[protocol.ConsensusCurrentVersion] protoParams.MaxBalLookback = 32 @@ -223,7 +223,7 @@ func TestBasicCatchpointWriter(t *testing.T) { } func TestFullCatchpointWriter(t *testing.T) { - // create new protocol version, which has lower back balance. + // create new protocol version, which has lower lookback testProtocolVersion := protocol.ConsensusVersion("test-protocol-TestFullCatchpointWriter") protoParams := config.Consensus[protocol.ConsensusCurrentVersion] protoParams.MaxBalLookback = 32 From 8112001c6944228dcf0ca7493061777128623d91 Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 10:33:28 -0500 Subject: [PATCH 04/10] Add benchmark --- ledger/acctupdates_test.go | 62 +++++++++++++++++++++++++++++++++++++- 1 file changed, 61 insertions(+), 1 deletion(-) diff --git a/ledger/acctupdates_test.go b/ledger/acctupdates_test.go index f40b6545fd..55a6aae902 100644 --- a/ledger/acctupdates_test.go +++ b/ledger/acctupdates_test.go @@ -554,7 +554,7 @@ func TestLargeAccountCountCatchpointGeneration(t *testing.T) { if runtime.GOARCH == "arm" || runtime.GOARCH == "arm64" { t.Skip("This test is too slow on ARM and causes travis builds to time out") } - // create new protocol version, which has lower back balance. + // create new protocol version, which has lower lookback testProtocolVersion := protocol.ConsensusVersion("test-protocol-TestLargeAccountCountCatchpointGeneration") protoParams := config.Consensus[protocol.ConsensusCurrentVersion] protoParams.MaxBalLookback = 32 @@ -1130,3 +1130,63 @@ func TestGetCatchpointStream(t *testing.T) { err = au.deleteStoredCatchpoints(context.Background(), au.accountsq) require.NoError(t, err) } + +func BenchmarkLargeMerkleTrieRebuild(b *testing.B) { + proto := config.Consensus[protocol.ConsensusCurrentVersion] + + ml := makeMockLedgerForTracker(b, true) + defer ml.close() + ml.blocks = randomInitChain(protocol.ConsensusCurrentVersion, 10) + + accts := []map[basics.Address]basics.AccountData{randomAccounts(5, true)} + + pooldata := basics.AccountData{} + pooldata.MicroAlgos.Raw = 1000 * 1000 * 1000 * 1000 + pooldata.Status = basics.NotParticipating + accts[0][testPoolAddr] = pooldata + + sinkdata := basics.AccountData{} + sinkdata.MicroAlgos.Raw = 1000 * 1000 * 1000 * 1000 + sinkdata.Status = basics.NotParticipating + accts[0][testSinkAddr] = sinkdata + + au := &accountUpdates{} + cfg := config.GetDefaultLocal() + cfg.Archival = true + au.initialize(cfg, ".", proto, accts[0]) + defer au.close() + + err := au.loadFromDisk(ml) + require.NoError(b, err) + + // at this point, the database was created. We want to fill the accounts data + accountsNumber := 6000000 * b.N + for i := 0; i < accountsNumber; { + updates := make(map[basics.Address]accountDelta, 0) + for k := 0; i < accountsNumber && k < 1024; k++ { + addr := randomAddress() + acctData := basics.AccountData{} + acctData.MicroAlgos.Raw = 1 + updates[addr] = accountDelta{new: acctData} + i++ + } + + err := ml.dbs.wdb.Atomic(func(ctx context.Context, tx *sql.Tx) (err error) { + return accountsNewRound(tx, updates, nil, proto) + }) + require.NoError(b, err) + } + + err = ml.dbs.wdb.Atomic(func(ctx context.Context, tx *sql.Tx) (err error) { + return updateAccountsRound(tx, 0, 1) + }) + require.NoError(b, err) + + au.close() + + b.ResetTimer() + err = au.loadFromDisk(ml) + require.NoError(b, err) + b.StopTimer() + b.ReportMetric(float64(accountsNumber), "entries/trie") +} From f488ab32a0ee15a3ba60bd6a3249a1fa6743c40b Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 11:05:52 -0500 Subject: [PATCH 05/10] Update comments. --- ledger/accountdb.go | 31 +++++++++++++++++++++++++------ ledger/acctupdates.go | 2 +- 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index c25e71f330..38bc18712d 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1132,8 +1132,8 @@ func (iterator *encodedAccountsBatchIter) Close() { } } -// orderedAccountsBuilderIter allows us to iterate over the accounts addresses in the order of the account hashes. -type orderedAccountsBuilderIter struct { +// orderedAccountsIter allows us to iterate over the accounts addresses in the order of the account hashes. +type orderedAccountsIter struct { step int rows *sql.Rows tx *sql.Tx @@ -1142,8 +1142,10 @@ type orderedAccountsBuilderIter struct { insertStmt *sql.Stmt } -func makeOrderedAccountsBuilderIter(tx *sql.Tx, accountCount int, accountData bool) *orderedAccountsBuilderIter { - return &orderedAccountsBuilderIter{ +// makeOrderedAccountsIter creates an ordered account iterator. Note that due to implementation reasons, +// only a single iterator can be active at a time. +func makeOrderedAccountsIter(tx *sql.Tx, accountCount int, accountData bool) *orderedAccountsIter { + return &orderedAccountsIter{ tx: tx, accountCount: accountCount, accountData: accountData, @@ -1151,14 +1153,25 @@ func makeOrderedAccountsBuilderIter(tx *sql.Tx, accountCount int, accountData bo } } +// accountAddressHashData is used by Next to return a single account address, hash and account data. type accountAddressHashData struct { address basics.Address digest []byte encodedAccountData []byte } -func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []accountAddressHashData, ordered bool, err error) { +// Next returns an array containing the account address, hash and potentially data +// the Next function works in multiple processing stages, where it first processs the current accounts and order them +// followed by returning the ordered accounts. In the first phase, it would return empty accountAddressHashData array +// whose size matches with the data that was processed, and set the ordered to false. On the second phase, the acct +// would contain valid data ( and optionally the account data as well, if was asked in makeOrderedAccountsIter) and +// would set the ordered to true. If err is sql.ErrNoRows it means that the iterator have completed it's work and no further +// accounts exists. Otherwise, the caller is expected to keep calling "Next" to retrieve the next set of accounts +// ( or let the Next function make some progress toward that goal ) +func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAddressHashData, ordered bool, err error) { if iterator.step == 0 { + // although we're going to delete this table anyway when completing the iterator execution, we'll try to + // clean up any intermediate table. _, err = iterator.tx.ExecContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") if err != nil { return @@ -1167,6 +1180,7 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac return } if iterator.step == 1 { + // create the temporary table _, err = iterator.tx.ExecContext(ctx, "CREATE TABLE accountsiteratorhashes(address blob, hash blob)") if err != nil { return @@ -1175,10 +1189,12 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac return } if iterator.step == 2 { + // iterate over the existing accounts iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT address, data FROM accountbase") if err != nil { return } + // prepare the insert statement into the temporary table iterator.insertStmt, err = iterator.tx.PrepareContext(ctx, "INSERT INTO accountsiteratorhashes(address, hash) VALUES(?, ?)") if err != nil { return @@ -1232,6 +1248,8 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac return } if iterator.step == 4 { + // create an index. It shown that even when we're making a single select statement in step 5, it would be better to have this index vs. not having it at all. + // note that this index is using the rowid of the accountsiteratorhashes table. _, err = iterator.tx.ExecContext(ctx, "CREATE INDEX accountsiteratorhashesidx ON accountsiteratorhashes(hash)") if err != nil { return @@ -1240,6 +1258,7 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac return } if iterator.step == 5 { + // select the data from the ordered table if iterator.accountData { iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT accountsiteratorhashes.address, accountsiteratorhashes.hash, accountbase.data FROM accountsiteratorhashes JOIN accountbase ON accountbase.address=accountsiteratorhashes.address ORDER BY accountsiteratorhashes.hash") } else { @@ -1302,7 +1321,7 @@ func (iterator *orderedAccountsBuilderIter) Next(ctx context.Context) (acct []ac } // Close shuts down the orderedAccountsBuilderIter, releasing database resources. -func (iterator *orderedAccountsBuilderIter) Close(ctx context.Context) (err error) { +func (iterator *orderedAccountsIter) Close(ctx context.Context) (err error) { if iterator.rows != nil { iterator.rows.Close() iterator.rows = nil diff --git a/ledger/acctupdates.go b/ledger/acctupdates.go index 3eb4849701..c517b85c2c 100644 --- a/ledger/acctupdates.go +++ b/ledger/acctupdates.go @@ -1119,7 +1119,7 @@ func (au *accountUpdates) accountsInitialize(ctx context.Context, tx *sql.Tx) (b if rootHash.IsZero() { au.log.Infof("accountsInitialize rebuilding merkle trie for round %d", rnd) - accountBuilderIt := makeOrderedAccountsBuilderIter(tx, trieRebuildAccountChunkSize, false) + accountBuilderIt := makeOrderedAccountsIter(tx, trieRebuildAccountChunkSize, false) defer accountBuilderIt.Close(ctx) startTrieBuildTime := time.Now() accountsCount := 0 From b1ea951e101cdccf00c8d18185b92d4421473b51 Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 16:18:47 -0500 Subject: [PATCH 06/10] update per reviewer request. --- ledger/accountdb.go | 84 +++++++++++++++++++++++++++++---------------- 1 file changed, 55 insertions(+), 29 deletions(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index 38bc18712d..db662eb2a8 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1132,24 +1132,50 @@ func (iterator *encodedAccountsBatchIter) Close() { } } +// orderedAccountsIterStep is used to by orderedAccountsIter to define the current step +type orderedAccountsIterStep int + +const ( + // startup step + oaiStepStartup = orderedAccountsIterStep(0) + // delete old ordering table if we have any leftover from previous invocation + oaiStepDeleteOldOrderingTable = orderedAccountsIterStep(0) + // create new ordering table + oaiStepCreateOrderingTable = orderedAccountsIterStep(1) + // query the existing accounts + oaiStepQueryAccounts = orderedAccountsIterStep(2) + // iterate over the existing accounts and insert their hash & address into the staging ordering table + oaiStepInsertAccountData = orderedAccountsIterStep(3) + // create an index on the ordering table so that we can efficiently scan it. + oaiStepCreateOrderingAccountIndex = orderedAccountsIterStep(4) + // query the ordering table + oaiStepSelectFromOrderedTable = orderedAccountsIterStep(5) + // iterate over the ordering table + oaiStepIterateOverOrderedTable = orderedAccountsIterStep(6) + // cleanup and delete ordering table + oaiStepShutdown = orderedAccountsIterStep(7) + // do nothing as we're done. + oaiStepDone = orderedAccountsIterStep(8) +) + // orderedAccountsIter allows us to iterate over the accounts addresses in the order of the account hashes. type orderedAccountsIter struct { - step int - rows *sql.Rows - tx *sql.Tx - accountCount int - accountData bool - insertStmt *sql.Stmt + step orderedAccountsIterStep + rows *sql.Rows + tx *sql.Tx + accountCount int + fetchAccountData bool + insertStmt *sql.Stmt } // makeOrderedAccountsIter creates an ordered account iterator. Note that due to implementation reasons, // only a single iterator can be active at a time. -func makeOrderedAccountsIter(tx *sql.Tx, accountCount int, accountData bool) *orderedAccountsIter { +func makeOrderedAccountsIter(tx *sql.Tx, accountCount int, fetchAccountData bool) *orderedAccountsIter { return &orderedAccountsIter{ - tx: tx, - accountCount: accountCount, - accountData: accountData, - step: 0, + tx: tx, + accountCount: accountCount, + fetchAccountData: fetchAccountData, + step: oaiStepStartup, } } @@ -1169,26 +1195,26 @@ type accountAddressHashData struct { // accounts exists. Otherwise, the caller is expected to keep calling "Next" to retrieve the next set of accounts // ( or let the Next function make some progress toward that goal ) func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAddressHashData, ordered bool, err error) { - if iterator.step == 0 { + if iterator.step == oaiStepDeleteOldOrderingTable { // although we're going to delete this table anyway when completing the iterator execution, we'll try to // clean up any intermediate table. _, err = iterator.tx.ExecContext(ctx, "DROP TABLE IF EXISTS accountsiteratorhashes") if err != nil { return } - iterator.step = 1 + iterator.step = oaiStepCreateOrderingTable return } - if iterator.step == 1 { + if iterator.step == oaiStepCreateOrderingTable { // create the temporary table _, err = iterator.tx.ExecContext(ctx, "CREATE TABLE accountsiteratorhashes(address blob, hash blob)") if err != nil { return } - iterator.step = 2 + iterator.step = oaiStepQueryAccounts return } - if iterator.step == 2 { + if iterator.step == oaiStepQueryAccounts { // iterate over the existing accounts iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT address, data FROM accountbase") if err != nil { @@ -1199,10 +1225,10 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd if err != nil { return } - iterator.step = 3 + iterator.step = oaiStepInsertAccountData return } - if iterator.step == 3 { + if iterator.step == oaiStepInsertAccountData { var addr basics.Address count := 0 for iterator.rows.Next() { @@ -1244,22 +1270,22 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd iterator.rows = nil iterator.insertStmt.Close() iterator.insertStmt = nil - iterator.step = 4 + iterator.step = oaiStepCreateOrderingAccountIndex return } - if iterator.step == 4 { + if iterator.step == oaiStepCreateOrderingAccountIndex { // create an index. It shown that even when we're making a single select statement in step 5, it would be better to have this index vs. not having it at all. // note that this index is using the rowid of the accountsiteratorhashes table. _, err = iterator.tx.ExecContext(ctx, "CREATE INDEX accountsiteratorhashesidx ON accountsiteratorhashes(hash)") if err != nil { return } - iterator.step = 5 + iterator.step = oaiStepSelectFromOrderedTable return } - if iterator.step == 5 { + if iterator.step == oaiStepSelectFromOrderedTable { // select the data from the ordered table - if iterator.accountData { + if iterator.fetchAccountData { iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT accountsiteratorhashes.address, accountsiteratorhashes.hash, accountbase.data FROM accountsiteratorhashes JOIN accountbase ON accountbase.address=accountsiteratorhashes.address ORDER BY accountsiteratorhashes.hash") } else { iterator.rows, err = iterator.tx.QueryContext(ctx, "SELECT address, hash FROM accountsiteratorhashes ORDER BY hash") @@ -1268,18 +1294,18 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd if err != nil { return } - iterator.step = 6 + iterator.step = oaiStepIterateOverOrderedTable return } - if iterator.step == 6 { + if iterator.step == oaiStepIterateOverOrderedTable { acct = make([]accountAddressHashData, 0, iterator.accountCount) var addr basics.Address for iterator.rows.Next() { var addrbuf []byte var acctdata []byte var hash []byte - if iterator.accountData { + if iterator.fetchAccountData { err = iterator.rows.Scan(&addrbuf, &hash, &acctdata) } else { err = iterator.rows.Scan(&addrbuf, &hash) @@ -1304,17 +1330,17 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd } } ordered = true - iterator.step = 7 + iterator.step = oaiStepShutdown iterator.rows.Close() iterator.rows = nil return } - if iterator.step == 7 { + if iterator.step == oaiStepShutdown { err = iterator.Close(ctx) if err != nil { return } - iterator.step = 8 + iterator.step = oaiStepDone // fallthrough } return nil, true, sql.ErrNoRows From faddefed9e8dbc517e9f472fa2d8c545e383c279 Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Fri, 13 Nov 2020 16:34:06 -0500 Subject: [PATCH 07/10] msgp ignore orderedAccountsIterStep --- ledger/accountdb.go | 1 + 1 file changed, 1 insertion(+) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index db662eb2a8..f956ea6d33 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1133,6 +1133,7 @@ func (iterator *encodedAccountsBatchIter) Close() { } // orderedAccountsIterStep is used to by orderedAccountsIter to define the current step +//msgp:ignore orderedAccountsIterStep type orderedAccountsIterStep int const ( From af2deddc158adf767d2697212c97994a8448bbcb Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Mon, 16 Nov 2020 16:27:39 -0500 Subject: [PATCH 08/10] fix typo --- ledger/accountdb.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index f956ea6d33..8b63d6135c 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1132,7 +1132,7 @@ func (iterator *encodedAccountsBatchIter) Close() { } } -// orderedAccountsIterStep is used to by orderedAccountsIter to define the current step +// orderedAccountsIterStep is used by orderedAccountsIter to define the current step //msgp:ignore orderedAccountsIterStep type orderedAccountsIterStep int From 78281f90ca85fae854a0aea9e9085315893467bb Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Tue, 17 Nov 2020 08:27:27 -0500 Subject: [PATCH 09/10] updating per reviewer's requested changes. --- ledger/accountdb.go | 19 +++++++----- ledger/acctupdates.go | 69 +++++++++++++++++++------------------------ 2 files changed, 42 insertions(+), 46 deletions(-) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index 8b63d6135c..c0f303d59a 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1190,12 +1190,12 @@ type accountAddressHashData struct { // Next returns an array containing the account address, hash and potentially data // the Next function works in multiple processing stages, where it first processs the current accounts and order them // followed by returning the ordered accounts. In the first phase, it would return empty accountAddressHashData array -// whose size matches with the data that was processed, and set the ordered to false. On the second phase, the acct +// and sets the processedRecords to the number of accounts that were processed. On the second phase, the acct // would contain valid data ( and optionally the account data as well, if was asked in makeOrderedAccountsIter) and -// would set the ordered to true. If err is sql.ErrNoRows it means that the iterator have completed it's work and no further +// the processedRecords would be zero. If err is sql.ErrNoRows it means that the iterator have completed it's work and no further // accounts exists. Otherwise, the caller is expected to keep calling "Next" to retrieve the next set of accounts // ( or let the Next function make some progress toward that goal ) -func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAddressHashData, ordered bool, err error) { +func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAddressHashData, processedRecords int, err error) { if iterator.step == oaiStepDeleteOldOrderingTable { // although we're going to delete this table anyway when completing the iterator execution, we'll try to // clean up any intermediate table. @@ -1243,6 +1243,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd if len(addrbuf) != len(addr) { err = fmt.Errorf("Account DB address length mismatch: %d != %d", len(addrbuf), len(addr)) + iterator.Close(ctx) return } @@ -1251,22 +1252,24 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd var accountData basics.AccountData err = protocol.Decode(buf, &accountData) if err != nil { + iterator.Close(ctx) return } hash := accountHashBuilder(addr, accountData, buf) _, err = iterator.insertStmt.ExecContext(ctx, addrbuf, hash) if err != nil { + iterator.Close(ctx) return } count++ if count == iterator.accountCount { // we're done with this iteration. - acct = make([]accountAddressHashData, count, count) + processedRecords = count return } } - acct = make([]accountAddressHashData, count, count) + processedRecords = count iterator.rows.Close() iterator.rows = nil iterator.insertStmt.Close() @@ -1279,6 +1282,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd // note that this index is using the rowid of the accountsiteratorhashes table. _, err = iterator.tx.ExecContext(ctx, "CREATE INDEX accountsiteratorhashesidx ON accountsiteratorhashes(hash)") if err != nil { + iterator.Close(ctx) return } iterator.step = oaiStepSelectFromOrderedTable @@ -1293,6 +1297,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd } if err != nil { + iterator.Close(ctx) return } iterator.step = oaiStepIterateOverOrderedTable @@ -1326,11 +1331,9 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd acct = append(acct, accountAddressHashData{address: addr, digest: hash, encodedAccountData: acctdata}) if len(acct) == iterator.accountCount { // we're done with this iteration. - ordered = true return } } - ordered = true iterator.step = oaiStepShutdown iterator.rows.Close() iterator.rows = nil @@ -1344,7 +1347,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd iterator.step = oaiStepDone // fallthrough } - return nil, true, sql.ErrNoRows + return nil, 0, sql.ErrNoRows } // Close shuts down the orderedAccountsBuilderIter, releasing database resources. diff --git a/ledger/acctupdates.go b/ledger/acctupdates.go index c517b85c2c..2a252351bf 100644 --- a/ledger/acctupdates.go +++ b/ledger/acctupdates.go @@ -1125,9 +1125,9 @@ func (au *accountUpdates) accountsInitialize(ctx context.Context, tx *sql.Tx) (b accountsCount := 0 lastRebuildTime := startTrieBuildTime pendingAccounts := 0 - wasOrdered := false + totalOrderedAccounts := 0 for { - accts, ordered, err := accountBuilderIt.Next(ctx) + accts, processedRows, err := accountBuilderIt.Next(ctx) if err == sql.ErrNoRows { // the account builder would return sql.ErrNoRows when no more data is available. break @@ -1135,48 +1135,41 @@ func (au *accountUpdates) accountsInitialize(ctx context.Context, tx *sql.Tx) (b return rnd, err } - if !wasOrdered && ordered { - accountsCount = 0 - wasOrdered = true - } - - accountsCount += len(accts) if len(accts) > 0 { - if ordered { - pendingAccounts += len(accts) - for _, acct := range accts { - added, err := trie.Add(acct.digest) - if err != nil { - return rnd, fmt.Errorf("accountsInitialize was unable to add changes to trie: %v", err) - } - if !added { - au.log.Warnf("accountsInitialize attempted to add duplicate hash '%s' to merkle trie for account %v", hex.EncodeToString(acct.digest), acct.address) - } + accountsCount += len(accts) + pendingAccounts += len(accts) + for _, acct := range accts { + added, err := trie.Add(acct.digest) + if err != nil { + return rnd, fmt.Errorf("accountsInitialize was unable to add changes to trie: %v", err) } - - if pendingAccounts >= trieRebuildCommitFrequency { - // this trie Evict will commit using the current transaction. - // if anything goes wrong, it will still get rolled back. - _, err = trie.Evict(true) - if err != nil { - return 0, fmt.Errorf("accountsInitialize was unable to commit changes to trie: %v", err) - } - pendingAccounts = 0 + if !added { + au.log.Warnf("accountsInitialize attempted to add duplicate hash '%s' to merkle trie for account %v", hex.EncodeToString(acct.digest), acct.address) } + } - if time.Now().Sub(lastRebuildTime) > 5*time.Second { - // let the user know that the trie is still being rebuilt. - au.log.Infof("accountsInitialize still building the trie, and processed so far %d accounts", accountsCount) - lastRebuildTime = time.Now() + if pendingAccounts >= trieRebuildCommitFrequency { + // this trie Evict will commit using the current transaction. + // if anything goes wrong, it will still get rolled back. + _, err = trie.Evict(true) + if err != nil { + return 0, fmt.Errorf("accountsInitialize was unable to commit changes to trie: %v", err) } + pendingAccounts = 0 + } - } else { - // if it's not ordered, we can ignore it for now; we'll just increase the counters and emit logs periodically. - if time.Now().Sub(lastRebuildTime) > 5*time.Second { - // let the user know that the trie is still being rebuilt. - au.log.Infof("accountsInitialize still building the trie, and hashed so far %d accounts", accountsCount) - lastRebuildTime = time.Now() - } + if time.Now().Sub(lastRebuildTime) > 5*time.Second { + // let the user know that the trie is still being rebuilt. + au.log.Infof("accountsInitialize still building the trie, and processed so far %d accounts", accountsCount) + lastRebuildTime = time.Now() + } + } else if processedRows > 0 { + totalOrderedAccounts += processedRows + // if it's not ordered, we can ignore it for now; we'll just increase the counters and emit logs periodically. + if time.Now().Sub(lastRebuildTime) > 5*time.Second { + // let the user know that the trie is still being rebuilt. + au.log.Infof("accountsInitialize still building the trie, and hashed so far %d accounts", totalOrderedAccounts) + lastRebuildTime = time.Now() } } } From 756fb461621378d039e1ac218dc70b8f2790f0ce Mon Sep 17 00:00:00 2001 From: Tsachi Herman Date: Tue, 17 Nov 2020 09:21:33 -0500 Subject: [PATCH 10/10] Add missing iterator.Close on error --- ledger/accountdb.go | 1 + 1 file changed, 1 insertion(+) diff --git a/ledger/accountdb.go b/ledger/accountdb.go index c0f303d59a..bd6b524e33 100644 --- a/ledger/accountdb.go +++ b/ledger/accountdb.go @@ -1323,6 +1323,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd if len(addrbuf) != len(addr) { err = fmt.Errorf("Account DB address length mismatch: %d != %d", len(addrbuf), len(addr)) + iterator.Close(ctx) return }