Merge a2e62140ac into cb5235eb07

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)
 	}

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
+	// Subscribe to the wallet feed and collect events.
-	updates := make(chan accounts.WalletEvent, 1)
+	var (
-	sub := ks.Subscribe(updates)
+		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
+	// Randomly add and remove accounts.
-	live := make(map[common.Address]accounts.Account)
+	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()
+	// Shut down the event collector and check events.
-		if len(liveList) != len(wallets) {
+	sub.Unsubscribe()
-			t.Errorf("wallet list doesn't match required accounts: have %v, want %v", wallets, liveList)
+	<-updates
-		} else {
+	checkAccounts(t, live, ks.Wallets())
-			for j, wallet := range wallets {
+	checkEvents(t, wantEvents, events)
-				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] {
+// checkAccounts checks that all known live accounts are present in the wallet list.
-					t.Errorf("wallet %d: account mismatch: have %v, want %v", j, accs[0], liveList[j])
+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)
 		}
 	}
 }

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)
+		debounce         = time.NewTimer(0)
-		debounceDuration  = 500 * time.Millisecond
+		debounceDuration = 500 * time.Millisecond
-		inCycle, hadEvent bool
+		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()
+			//We're now handling the events, scan again as long as new
-			w.ac.reload()
+			// events keep coming during our fs-scan
-			w.ac.mu.Unlock()
+			var (
-			if hadEvent {
+				accs []accounts.Account
-				debounce.Reset(debounceDuration)
+				err  error
-				inCycle, hadEvent = true, false
+			)
-			} else {
+			// Scan again if more events occurred during scan
-				inCycle, hadEvent = false, false
+			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)
 		}
 	}
 }