diff --git a/accounts/keystore/account_cache.go b/accounts/keystore/account_cache.go index dc6ac6ccb3..77c19ddd9f 100644 --- a/accounts/keystore/account_cache.go +++ b/accounts/keystore/account_cache.go @@ -209,6 +209,12 @@ func (ac *accountCache) close() { // Callers must hold ac.mu. func (ac *accountCache) reload() { accounts, err := ac.scan() + ac.handleScanResult(accounts, err) +} + +// handleScanResult uses the result of a fs scan to update the account lists +// This can be used to perform un-mutexed fs scans +func (ac *accountCache) handleScanResult(accounts []accounts.Account, err error) { if err != nil { log.Debug("Failed to reload keystore contents", "err", err) } diff --git a/accounts/keystore/keystore_test.go b/accounts/keystore/keystore_test.go index 132b74439e..6fb0a78088 100644 --- a/accounts/keystore/keystore_test.go +++ b/accounts/keystore/keystore_test.go @@ -272,82 +272,104 @@ func TestWalletNotifierLifecycle(t *testing.T) { t.Errorf("wallet notifier didn't terminate after unsubscribe") } +type walletEvent struct { + accounts.WalletEvent + a accounts.Account +} + // Tests that wallet notifications and correctly fired when accounts are added // or deleted from the keystore. func TestWalletNotifications(t *testing.T) { - // Create a temporary kesytore to test with dir, ks := tmpKeyStore(t, false) defer os.RemoveAll(dir) - // Subscribe to the wallet feed - updates := make(chan accounts.WalletEvent, 1) - sub := ks.Subscribe(updates) + // Subscribe to the wallet feed and collect events. + var ( + events []walletEvent + updates = make(chan accounts.WalletEvent) + sub = ks.Subscribe(updates) + ) defer sub.Unsubscribe() + go func() { + for { + select { + case ev := <-updates: + events = append(events, walletEvent{ev, ev.Wallet.Accounts()[0]}) + case <-sub.Err(): + close(updates) + return + } + } + }() - // Randomly add and remove account and make sure events and wallets are in sync - live := make(map[common.Address]accounts.Account) + // Randomly add and remove accounts. + var ( + live = make(map[common.Address]accounts.Account) + wantEvents []walletEvent + ) for i := 0; i < 1024; i++ { - // Execute a creation or deletion and ensure event arrival if create := len(live) == 0 || rand.Int()%4 > 0; create { // Add a new account and ensure wallet notifications arrives account, err := ks.NewAccount("") if err != nil { t.Fatalf("failed to create test account: %v", err) } - select { - case event := <-updates: - if event.Kind != accounts.WalletArrived { - t.Errorf("non-arrival event on account creation") - } - if event.Wallet.Accounts()[0] != account { - t.Errorf("account mismatch on created wallet: have %v, want %v", event.Wallet.Accounts()[0], account) - } - default: - t.Errorf("wallet arrival event not fired on account creation") - } live[account.Address] = account + wantEvents = append(wantEvents, walletEvent{accounts.WalletEvent{Kind: accounts.WalletArrived}, account}) } else { - // Select a random account to delete (crude, but works) + // Delete a random account. var account accounts.Account for _, a := range live { account = a break } - // Remove an account and ensure wallet notifiaction arrives if err := ks.Delete(account, ""); err != nil { t.Fatalf("failed to delete test account: %v", err) } - select { - case event := <-updates: - if event.Kind != accounts.WalletDropped { - t.Errorf("non-drop event on account deletion") - } - if event.Wallet.Accounts()[0] != account { - t.Errorf("account mismatch on deleted wallet: have %v, want %v", event.Wallet.Accounts()[0], account) - } - default: - t.Errorf("wallet departure event not fired on account creation") - } delete(live, account.Address) + wantEvents = append(wantEvents, walletEvent{accounts.WalletEvent{Kind: accounts.WalletDropped}, account}) } - // Retrieve the list of wallets and ensure it matches with our required live set - liveList := make([]accounts.Account, 0, len(live)) - for _, account := range live { - liveList = append(liveList, account) - } - sort.Sort(accountsByURL(liveList)) + } - wallets := ks.Wallets() - if len(liveList) != len(wallets) { - t.Errorf("wallet list doesn't match required accounts: have %v, want %v", wallets, liveList) - } else { - for j, wallet := range wallets { - if accs := wallet.Accounts(); len(accs) != 1 { - t.Errorf("wallet %d: contains invalid number of accounts: have %d, want 1", j, len(accs)) - } else if accs[0] != liveList[j] { - t.Errorf("wallet %d: account mismatch: have %v, want %v", j, accs[0], liveList[j]) - } + // Shut down the event collector and check events. + sub.Unsubscribe() + <-updates + checkAccounts(t, live, ks.Wallets()) + checkEvents(t, wantEvents, events) +} + +// checkAccounts checks that all known live accounts are present in the wallet list. +func checkAccounts(t *testing.T, live map[common.Address]accounts.Account, wallets []accounts.Wallet) { + if len(live) != len(wallets) { + t.Errorf("wallet list doesn't match required accounts: have %d, want %d", len(wallets), len(live)) + return + } + liveList := make([]accounts.Account, 0, len(live)) + for _, account := range live { + liveList = append(liveList, account) + } + sort.Sort(accountsByURL(liveList)) + for j, wallet := range wallets { + if accs := wallet.Accounts(); len(accs) != 1 { + t.Errorf("wallet %d: contains invalid number of accounts: have %d, want 1", j, len(accs)) + } else if accs[0] != liveList[j] { + t.Errorf("wallet %d: account mismatch: have %v, want %v", j, accs[0], liveList[j]) + } + } +} + +// checkEvents checks that all events in 'want' are present in 'have'. Events may be present multiple times. +func checkEvents(t *testing.T, want []walletEvent, have []walletEvent) { + for _, wantEv := range want { + nmatch := 0 + for ; len(have) > 0; nmatch++ { + if have[0].Kind != wantEv.Kind || have[0].a != wantEv.a { + break } + have = have[1:] + } + if nmatch == 0 { + t.Fatalf("can't find event with Kind=%v for %x", wantEv.Kind, wantEv.a.Address) } } } diff --git a/accounts/keystore/watch.go b/accounts/keystore/watch.go index f4d6477912..25bdf44a62 100644 --- a/accounts/keystore/watch.go +++ b/accounts/keystore/watch.go @@ -21,8 +21,10 @@ package keystore import ( "time" + "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/log" "github.com/rjeczalik/notify" + "sync/atomic" ) type watcher struct { @@ -82,9 +84,10 @@ func (w *watcher) loop() { // When an event occurs, the reload call is delayed a bit so that // multiple events arriving quickly only cause a single reload. var ( - debounce = time.NewTimer(0) - debounceDuration = 500 * time.Millisecond - inCycle, hadEvent bool + debounce = time.NewTimer(0) + debounceDuration = 500 * time.Millisecond + unHandledEvents = uint64(0) + inCycle = uint64(0) ) defer debounce.Stop() for { @@ -92,22 +95,28 @@ func (w *watcher) loop() { case <-w.quit: return case <-w.ev: - if !inCycle { + // Count up the unhandled events + atomic.AddUint64(&unHandledEvents, 1) + // Trigger the scan (with delay), if not already triggered + if atomic.SwapUint64(&inCycle, 1) == 0 { debounce.Reset(debounceDuration) - inCycle = true - } else { - hadEvent = true } case <-debounce.C: - w.ac.mu.Lock() - w.ac.reload() - w.ac.mu.Unlock() - if hadEvent { - debounce.Reset(debounceDuration) - inCycle, hadEvent = true, false - } else { - inCycle, hadEvent = false, false + //We're now handling the events, scan again as long as new + // events keep coming during our fs-scan + var ( + accs []accounts.Account + err error + ) + // Scan again if more events occurred during scan + for atomic.SwapUint64(&unHandledEvents, 0) > 0 { + accs, err = w.ac.scan() } + w.ac.mu.Lock() + w.ac.handleScanResult(accs, err) + w.ac.mu.Unlock() + // Signal we're finished with cycle + atomic.SwapUint64(&inCycle, 0) } } }