Fix hang related to account file writes.

The disk syncer now maintains its own countdown timer, creating a new
timer only when necessary (when there is no timer running, and
something is scheduled to be written).  When the timer expires, the
select loop begins selecting on a grab of the account manager's binary
semaphore, and if read, performs the sync and nils the select channel
to prevent a future grab until a new timer has expired.

Tested with a race-enabled build on Windows.  No lockups or races
related to the disk syncing experienced with constant client requests
and incoming btcd notifications, and scheduled writes run as expected
once the countdown timer expires, locking out all server request and
notifiation handling.
This commit is contained in:
Josh Rickmar 2014-02-04 15:48:20 -05:00
parent db576ba636
commit 114bb581f7
3 changed files with 37 additions and 30 deletions

View file

@ -25,7 +25,6 @@ import (
"github.com/conformal/btcwallet/tx" "github.com/conformal/btcwallet/tx"
"github.com/conformal/btcwallet/wallet" "github.com/conformal/btcwallet/wallet"
"github.com/conformal/btcwire" "github.com/conformal/btcwire"
"time"
) )
// Errors relating to accounts. // Errors relating to accounts.
@ -81,8 +80,6 @@ func (am *AccountManager) Start() {
l := list.New() l := list.New()
m := make(map[string]*Account) m := make(map[string]*Account)
wait := 10 * time.Second
timer := time.NewTimer(wait)
for { for {
select { select {
case access := <-am.accessAccount: case access := <-am.accessAccount:
@ -117,20 +114,18 @@ func (am *AccountManager) Start() {
} }
} }
} }
case <-timer.C:
am.ds.FlushScheduled()
timer = time.NewTimer(wait)
} }
} }
} }
// Grab grabs the account manager's binary semaphore. // Grab grabs the account manager's binary semaphore. A custom semaphore
// is used instead of a sync.Mutex so the account manager's disk syncer
// can grab the semaphore from a select statement.
func (am *AccountManager) Grab() { func (am *AccountManager) Grab() {
<-am.bsem <-am.bsem
} }
// Release releases the account manager's binary semaphore. // Release releases exclusive ownership of the AccountManager.
func (am *AccountManager) Release() { func (am *AccountManager) Release() {
am.bsem <- struct{}{} am.bsem <- struct{}{}
} }
@ -476,7 +471,7 @@ func (am *AccountManager) ListSinceBlock(since, curBlockHeight int32, minconf in
// GetTransaction(). // GetTransaction().
type accountTx struct { type accountTx struct {
Account string Account string
Tx tx.Tx Tx tx.Tx
} }
// GetTransaction returns an array of accountTx to fully represent the effect of // GetTransaction returns an array of accountTx to fully represent the effect of

View file

@ -22,6 +22,7 @@ import (
"io/ioutil" "io/ioutil"
"os" "os"
"path/filepath" "path/filepath"
"time"
) )
// networkDir returns the directory name of a network directory to hold account // networkDir returns the directory name of a network directory to hold account
@ -192,8 +193,7 @@ type exportRequest struct {
// DiskSyncer manages all disk write operations for a collection of accounts. // DiskSyncer manages all disk write operations for a collection of accounts.
type DiskSyncer struct { type DiskSyncer struct {
// Flush scheduled account writes. // Flush scheduled account writes.
flushScheduled chan struct{} flushAccount chan *flushAccountRequest
flushAccount chan *flushAccountRequest
// Schedule file writes for an account. // Schedule file writes for an account.
scheduleWallet chan *Account scheduleWallet chan *Account
@ -214,7 +214,6 @@ type DiskSyncer struct {
// NewDiskSyncer creates a new DiskSyncer. // NewDiskSyncer creates a new DiskSyncer.
func NewDiskSyncer(am *AccountManager) *DiskSyncer { func NewDiskSyncer(am *AccountManager) *DiskSyncer {
return &DiskSyncer{ return &DiskSyncer{
flushScheduled: make(chan struct{}, 1),
flushAccount: make(chan *flushAccountRequest), flushAccount: make(chan *flushAccountRequest),
scheduleWallet: make(chan *Account), scheduleWallet: make(chan *Account),
scheduleTxStore: make(chan *Account), scheduleTxStore: make(chan *Account),
@ -237,28 +236,52 @@ func (ds *DiskSyncer) Start() {
} }
tmpnetdir := tmpNetworkDir(cfg.Net()) tmpnetdir := tmpNetworkDir(cfg.Net())
const wait = 10 * time.Second
var timer <-chan time.Time
var sem chan struct{}
schedule := newSyncSchedule(netdir) schedule := newSyncSchedule(netdir)
for { for {
select { select {
case <-ds.flushScheduled: case <-sem: // Now have exclusive access of the account manager
ds.am.Grab()
err := schedule.flush() err := schedule.flush()
ds.am.Release()
if err != nil { if err != nil {
log.Errorf("Cannot write accounts: %v", err) log.Errorf("Cannot write accounts: %v", err)
} }
timer = nil
// Account manager passed ownership of the semaphore;
// Do not grab semaphore again until another flush is needed.
sem = nil
// Release semaphore.
ds.am.bsem <- struct{}{}
case <-timer:
// Grab AccountManager semaphore when ready so flush can occur.
sem = ds.am.bsem
case fr := <-ds.flushAccount: case fr := <-ds.flushAccount:
fr.err <- schedule.flushAccount(fr.a) fr.err <- schedule.flushAccount(fr.a)
case a := <-ds.scheduleWallet: case a := <-ds.scheduleWallet:
schedule.wallets[a] = struct{}{} schedule.wallets[a] = struct{}{}
if timer == nil {
timer = time.After(wait)
}
case a := <-ds.scheduleTxStore: case a := <-ds.scheduleTxStore:
schedule.txs[a] = struct{}{} schedule.txs[a] = struct{}{}
if timer == nil {
timer = time.After(wait)
}
case a := <-ds.scheduleUtxoStore: case a := <-ds.scheduleUtxoStore:
schedule.utxos[a] = struct{}{} schedule.utxos[a] = struct{}{}
if timer == nil {
timer = time.After(wait)
}
case sr := <-ds.writeBatch: case sr := <-ds.writeBatch:
err := batchWriteAccounts(sr.a, tmpnetdir, netdir) err := batchWriteAccounts(sr.a, tmpnetdir, netdir)
@ -266,6 +289,7 @@ func (ds *DiskSyncer) Start() {
// All accounts have been synced, old schedule // All accounts have been synced, old schedule
// can be discarded. // can be discarded.
schedule = newSyncSchedule(netdir) schedule = newSyncSchedule(netdir)
timer = nil
} }
sr.err <- err sr.err <- err
@ -277,17 +301,6 @@ func (ds *DiskSyncer) Start() {
} }
} }
// FlushScheduled writes all scheduled account files to disk.
func (ds *DiskSyncer) FlushScheduled() {
// Schedule a flush if one is not already waiting. This channel
// is buffered so if a request is already waiting, a duplicate
// can be safely dropped.
select {
case ds.flushScheduled <- struct{}{}:
default:
}
}
// FlushAccount writes all scheduled account files to disk for a single // FlushAccount writes all scheduled account files to disk for a single
// account. // account.
func (ds *DiskSyncer) FlushAccount(a *Account) error { func (ds *DiskSyncer) FlushAccount(a *Account) error {

View file

@ -789,7 +789,7 @@ func GetTransaction(icmd btcjson.Cmd) (interface{}, *btcjson.Error) {
return nil, &btcjson.ErrInternal return nil, &btcjson.ErrInternal
} }
accumulatedTxen := AcctMgr.GetTransaction(cmd.Txid) accumulatedTxen := AcctMgr.GetTransaction(cmd.Txid)
if len(accumulatedTxen) == 0 { if len(accumulatedTxen) == 0 {
return nil, &btcjson.ErrNoTxInfo return nil, &btcjson.ErrNoTxInfo
} }
@ -825,7 +825,7 @@ func GetTransaction(icmd btcjson.Cmd) (interface{}, *btcjson.Error) {
// whether it is an orphan or in the blockchain. // whether it is an orphan or in the blockchain.
"category": "receive", "category": "receive",
"amount": t.Amount, "amount": t.Amount,
"address": hex.EncodeToString(t.ReceiverHash), "address": hex.EncodeToString(t.ReceiverHash),
}) })
} }
} }
@ -1153,7 +1153,6 @@ func sendPairs(icmd btcjson.Cmd, account string, amounts map[string]int64,
return handleSendRawTxReply(icmd, txid, a, createdTx) return handleSendRawTxReply(icmd, txid, a, createdTx)
} }
// SendFrom handles a sendfrom RPC request by creating a new transaction // SendFrom handles a sendfrom RPC request by creating a new transaction
// spending unspent transaction outputs for a wallet to another payment // spending unspent transaction outputs for a wallet to another payment
// address. Leftover inputs not sent to the payment address or a fee for // address. Leftover inputs not sent to the payment address or a fee for