Synchronize all account file writes.
Fixes several hangs cased by incorrect locking, by removing the locking. Instead, a single goroutine manages all file writes. The old account 'dirty' boolean flags have been removed. Instead, anytime an account structure is modified, the portion that was modified (wallet, tx store, or utxo store) must be scheduled to be written.
This commit is contained in:
Josh Rickmar
parent
0b371b09e8
commit
430db140ee
6 changed files with 412 additions and 396 deletions
553
disksync.go
553
disksync.go
|
|
@ -22,20 +22,9 @@ import (
|
|||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
// dirtyAccounts holds a set of accounts that include dirty components.
|
||||
dirtyAccounts = struct {
|
||||
sync.Mutex
|
||||
m map[*Account]bool
|
||||
}{
|
||||
m: make(map[*Account]bool),
|
||||
}
|
||||
)
|
||||
|
||||
// networkDir returns the directory name of a network directory to hold account
|
||||
// files.
|
||||
func networkDir(net btcwire.BitcoinNet) string {
|
||||
|
|
@ -53,6 +42,33 @@ func tmpNetworkDir(net btcwire.BitcoinNet) string {
|
|||
return networkDir(net) + "_tmp"
|
||||
}
|
||||
|
||||
// freshDir creates a new directory specified by path if it does not
|
||||
// exist. If the directory already exists, all files contained in the
|
||||
// directory are removed.
|
||||
func freshDir(path string) error {
|
||||
if err := checkCreateDir(path); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Remove all files in the directory.
|
||||
fd, err := os.Open(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer fd.Close()
|
||||
names, err := fd.Readdirnames(0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, name := range names {
|
||||
if err := os.RemoveAll(name); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// checkCreateDir checks that the path exists and is a directory.
|
||||
// If path does not exist, it is created.
|
||||
func checkCreateDir(path string) error {
|
||||
|
|
@ -87,259 +103,320 @@ func accountFilename(suffix, account, netdir string) string {
|
|||
return filepath.Join(netdir, fmt.Sprintf("%v-%v", account, suffix))
|
||||
}
|
||||
|
||||
// DirtyAccountSyncer synces dirty accounts for cases where the updated
|
||||
// information was not required to be immediately written to disk. Accounts
|
||||
// may be added to dirtyAccounts and will be checked and processed every 10
|
||||
// seconds by this function.
|
||||
// syncSchedule references the account files which have been
|
||||
// scheduled to be written and the directory to write to.
|
||||
type syncSchedule struct {
|
||||
dir string
|
||||
wallets map[*Account]struct{}
|
||||
txs map[*Account]struct{}
|
||||
utxos map[*Account]struct{}
|
||||
}
|
||||
|
||||
func newSyncSchedule(dir string) *syncSchedule {
|
||||
s := &syncSchedule{
|
||||
dir: dir,
|
||||
wallets: make(map[*Account]struct{}),
|
||||
txs: make(map[*Account]struct{}),
|
||||
utxos: make(map[*Account]struct{}),
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// FlushAccount writes all scheduled account files to disk for
|
||||
// a single account and removes them from the schedule.
|
||||
func (s *syncSchedule) FlushAccount(a *Account) error {
|
||||
if _, ok := s.utxos[a]; ok {
|
||||
if err := a.writeUtxoStore(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.utxos, a)
|
||||
}
|
||||
if _, ok := s.txs[a]; ok {
|
||||
if err := a.writeTxStore(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.txs, a)
|
||||
}
|
||||
if _, ok := s.wallets[a]; ok {
|
||||
if err := a.writeWallet(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.wallets, a)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Flush writes all scheduled account files and removes each
|
||||
// from the schedule.
|
||||
func (s *syncSchedule) Flush() error {
|
||||
for a := range s.utxos {
|
||||
if err := a.writeUtxoStore(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.utxos, a)
|
||||
}
|
||||
|
||||
for a := range s.txs {
|
||||
if err := a.writeTxStore(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.txs, a)
|
||||
}
|
||||
|
||||
for a := range s.wallets {
|
||||
if err := a.writeWallet(s.dir); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(s.wallets, a)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Channels for AccountDiskSyncer.
|
||||
var (
|
||||
scheduleWalletWrite = make(chan *Account)
|
||||
scheduleTxStoreWrite = make(chan *Account)
|
||||
scheduleUtxoStoreWrite = make(chan *Account)
|
||||
syncBatch = make(chan *syncBatchRequest)
|
||||
syncAccount = make(chan *syncRequest)
|
||||
exportAccount = make(chan *exportRequest)
|
||||
)
|
||||
|
||||
type syncRequest struct {
|
||||
a *Account
|
||||
err chan error
|
||||
}
|
||||
|
||||
type syncBatchRequest struct {
|
||||
a []*Account
|
||||
err chan error
|
||||
}
|
||||
|
||||
type exportRequest struct {
|
||||
dir string
|
||||
a *Account
|
||||
err chan error
|
||||
}
|
||||
|
||||
// AccountDiskSyncer manages a set of "dirty" account files which must
|
||||
// be written to disk, and synchronizes all writes in a single goroutine.
|
||||
// After 10 seconds since the latest sync, all unwritten files are written
|
||||
// and removed. Writes for a single account may be scheduled immediately by
|
||||
// calling WriteScheduledToDisk.
|
||||
//
|
||||
// This never returns and is meant to be called from a goroutine.
|
||||
func DirtyAccountSyncer() {
|
||||
func AccountDiskSyncer() {
|
||||
netdir := networkDir(cfg.Net())
|
||||
if err := checkCreateDir(netdir); err != nil {
|
||||
log.Errorf("Unable to create or write to account directory: %v", err)
|
||||
}
|
||||
tmpnetdir := tmpNetworkDir(cfg.Net())
|
||||
|
||||
schedule := newSyncSchedule(netdir)
|
||||
ticker := time.Tick(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-ticker:
|
||||
dirtyAccounts.Lock()
|
||||
for a := range dirtyAccounts.m {
|
||||
log.Debugf("Syncing account '%v' to disk",
|
||||
a.Wallet.Name())
|
||||
if err := a.writeDirtyToDisk(); err != nil {
|
||||
log.Errorf("cannot sync dirty wallet: %v",
|
||||
err)
|
||||
} else {
|
||||
delete(dirtyAccounts.m, a)
|
||||
}
|
||||
case a := <-scheduleWalletWrite:
|
||||
schedule.wallets[a] = struct{}{}
|
||||
|
||||
case a := <-scheduleTxStoreWrite:
|
||||
schedule.txs[a] = struct{}{}
|
||||
|
||||
case a := <-scheduleUtxoStoreWrite:
|
||||
schedule.utxos[a] = struct{}{}
|
||||
|
||||
case sr := <-syncAccount:
|
||||
sr.err <- schedule.FlushAccount(sr.a)
|
||||
|
||||
case sr := <-syncBatch:
|
||||
err := batchWriteAccounts(sr.a, tmpnetdir, netdir)
|
||||
if err == nil {
|
||||
// All accounts have been synced, old schedule
|
||||
// can be discarded.
|
||||
schedule = newSyncSchedule(netdir)
|
||||
}
|
||||
sr.err <- err
|
||||
|
||||
case er := <-exportAccount:
|
||||
a := er.a
|
||||
dir := er.dir
|
||||
er.err <- a.writeAll(dir)
|
||||
|
||||
case <-ticker:
|
||||
if err := schedule.Flush(); err != nil {
|
||||
log.Errorf("Cannot write account: %v", err)
|
||||
}
|
||||
dirtyAccounts.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// freshDir creates a new directory specified by path if it does not
|
||||
// exist. If the directory already exists, all files contained in the
|
||||
// directory are removed.
|
||||
func freshDir(path string) error {
|
||||
if err := checkCreateDir(path); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Remove all files in the directory.
|
||||
fd, err := os.Open(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer fd.Close()
|
||||
names, err := fd.Readdirnames(0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, name := range names {
|
||||
if err := os.RemoveAll(name); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeAllToFreshDir writes all account files to the specified directory.
|
||||
// If dir already exists, any old files are removed. If dir does not
|
||||
// exist, it is created.
|
||||
// WriteAllToDisk writes all account files for all accounts at once. Unlike
|
||||
// writing individual account files, this causes each account file to be
|
||||
// written to a new network directory to replace the old one. Use this
|
||||
// function when it is needed to ensure an all or nothing write for all
|
||||
// account files.
|
||||
//
|
||||
// It is a runtime error to call this function while not holding each
|
||||
// wallet, tx store, and utxo store writer lock.
|
||||
func (a *Account) writeAllToFreshDir(dir string) error {
|
||||
if err := freshDir(dir); err != nil {
|
||||
return err
|
||||
// It is a runtime error to call this without holding the store writer lock.
|
||||
func (store *AccountStore) WriteAllToDisk() error {
|
||||
accts := make([]*Account, 0, len(store.accounts))
|
||||
for _, a := range store.accounts {
|
||||
accts = append(accts, a)
|
||||
}
|
||||
|
||||
err := make(chan error, 1)
|
||||
syncBatch <- &syncBatchRequest{
|
||||
a: accts,
|
||||
err: err,
|
||||
}
|
||||
return <-err
|
||||
}
|
||||
|
||||
func batchWriteAccounts(accts []*Account, tmpdir, netdir string) error {
|
||||
if err := freshDir(tmpdir); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, a := range accts {
|
||||
if err := a.writeAll(tmpdir); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// This is technically NOT an atomic operation, but at startup, if the
|
||||
// network directory is missing but the temporary network directory
|
||||
// exists, the temporary is moved before accounts are opened.
|
||||
if err := os.RemoveAll(netdir); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := Rename(tmpdir, netdir); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteScheduledToDisk signals AccountDiskSyncer to write all scheduled
|
||||
// account files for a to disk now instead of waiting for the next sync
|
||||
// interval. This function blocks until all the file writes for a have
|
||||
// finished, and returns a non-nil error if any of the file writes failed.
|
||||
func (a *Account) WriteScheduledToDisk() error {
|
||||
err := make(chan error, 1)
|
||||
syncAccount <- &syncRequest{
|
||||
a: a,
|
||||
err: err,
|
||||
}
|
||||
return <-err
|
||||
}
|
||||
|
||||
// ScheduleWalletWrite schedules a write of an account's wallet file.
|
||||
func (a *Account) ScheduleWalletWrite() {
|
||||
scheduleWalletWrite <- a
|
||||
}
|
||||
|
||||
// ScheduleTxStoreWrite schedules a write of an account's tx store file.
|
||||
func (a *Account) ScheduleTxStoreWrite() {
|
||||
scheduleTxStoreWrite <- a
|
||||
}
|
||||
|
||||
// ScheduleUtxoStoreWrite schedules a write of an account's utxo store file.
|
||||
func (a *Account) ScheduleUtxoStoreWrite() {
|
||||
scheduleUtxoStoreWrite <- a
|
||||
}
|
||||
|
||||
// ExportToDirectory writes an account to a special export directory. Any
|
||||
// previous files are overwritten.
|
||||
func (a *Account) ExportToDirectory(dirBaseName string) error {
|
||||
dir := filepath.Join(networkDir(cfg.Net()), dirBaseName)
|
||||
if err := checkCreateDir(dir); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Println("exporting to %v", dir)
|
||||
|
||||
err := make(chan error)
|
||||
er := &exportRequest{
|
||||
dir: dir,
|
||||
a: a,
|
||||
err: err,
|
||||
}
|
||||
exportAccount <- er
|
||||
return <-err
|
||||
}
|
||||
|
||||
func (a *Account) writeAll(dir string) error {
|
||||
if err := a.writeUtxoStore(dir); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.writeTxStore(dir); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.writeWallet(dir); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Account) writeWallet(dir string) error {
|
||||
wfilepath := accountFilename("wallet.bin", a.name, dir)
|
||||
_, filename := filepath.Split(wfilepath)
|
||||
tmpfile, err := ioutil.TempFile(dir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tmpfile.Close()
|
||||
|
||||
a.mtx.RLock()
|
||||
_, err = a.Wallet.WriteTo(tmpfile)
|
||||
a.mtx.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = Rename(tmpfile.Name(), wfilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Account) writeTxStore(dir string) error {
|
||||
txfilepath := accountFilename("tx.bin", a.name, dir)
|
||||
_, filename := filepath.Split(txfilepath)
|
||||
tmpfile, err := ioutil.TempFile(dir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tmpfile.Close()
|
||||
|
||||
a.TxStore.RLock()
|
||||
_, err = a.TxStore.s.WriteTo(tmpfile)
|
||||
a.TxStore.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = Rename(tmpfile.Name(), txfilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Account) writeUtxoStore(dir string) error {
|
||||
utxofilepath := accountFilename("utxo.bin", a.name, dir)
|
||||
|
||||
wfile, err := os.Create(wfilepath)
|
||||
_, filename := filepath.Split(utxofilepath)
|
||||
tmpfile, err := ioutil.TempFile(dir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer wfile.Close()
|
||||
txfile, err := os.Create(txfilepath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer txfile.Close()
|
||||
utxofile, err := os.Create(utxofilepath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer utxofile.Close()
|
||||
defer tmpfile.Close()
|
||||
|
||||
if _, err := a.Wallet.WriteTo(wfile); err != nil {
|
||||
return err
|
||||
}
|
||||
a.dirty = false
|
||||
|
||||
if _, err := a.TxStore.s.WriteTo(txfile); err != nil {
|
||||
return err
|
||||
}
|
||||
a.TxStore.dirty = false
|
||||
|
||||
if _, err := a.UtxoStore.s.WriteTo(utxofile); err != nil {
|
||||
return err
|
||||
}
|
||||
a.UtxoStore.dirty = false
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeDirtyToDisk checks for the dirty flag on an account's wallet,
|
||||
// txstore, and utxostore, writing them to disk if any are dirty.
|
||||
func (a *Account) writeDirtyToDisk() error {
|
||||
netdir := networkDir(cfg.Net())
|
||||
if err := checkCreateDir(netdir); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
wfilepath := accountFilename("wallet.bin", a.name, netdir)
|
||||
txfilepath := accountFilename("tx.bin", a.name, netdir)
|
||||
utxofilepath := accountFilename("utxo.bin", a.name, netdir)
|
||||
|
||||
// UTXOs and transactions are synced to disk first. This prevents
|
||||
// any races from saving a wallet marked to be synced with block N
|
||||
// and btcwallet closing while the UTXO and Tx files are only synced
|
||||
// with block N-1.
|
||||
|
||||
// UTXOs
|
||||
a.UtxoStore.RLock()
|
||||
dirty := a.UtxoStore.dirty
|
||||
a.UtxoStore.RUnlock()
|
||||
if dirty {
|
||||
netdir, filename := filepath.Split(utxofilepath)
|
||||
tmpfile, err := ioutil.TempFile(netdir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tmpfile.Close()
|
||||
|
||||
a.UtxoStore.RLock()
|
||||
_, err = a.UtxoStore.s.WriteTo(tmpfile)
|
||||
a.UtxoStore.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = Rename(tmpfile.Name(), utxofilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.UtxoStore.Lock()
|
||||
a.UtxoStore.dirty = false
|
||||
a.UtxoStore.Unlock()
|
||||
}
|
||||
|
||||
// Transactions
|
||||
a.TxStore.RLock()
|
||||
dirty = a.TxStore.dirty
|
||||
a.TxStore.RUnlock()
|
||||
if dirty {
|
||||
netdir, filename := filepath.Split(txfilepath)
|
||||
tmpfile, err := ioutil.TempFile(netdir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tmpfile.Close()
|
||||
|
||||
a.TxStore.RLock()
|
||||
_, err = a.TxStore.s.WriteTo(tmpfile)
|
||||
a.TxStore.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = Rename(tmpfile.Name(), txfilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.TxStore.Lock()
|
||||
a.TxStore.dirty = false
|
||||
a.TxStore.Unlock()
|
||||
}
|
||||
|
||||
// Wallet
|
||||
a.mtx.RLock()
|
||||
dirty = a.dirty
|
||||
a.mtx.RUnlock()
|
||||
if dirty {
|
||||
netdir, filename := filepath.Split(wfilepath)
|
||||
tmpfile, err := ioutil.TempFile(netdir, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tmpfile.Close()
|
||||
|
||||
a.mtx.RLock()
|
||||
_, err = a.Wallet.WriteTo(tmpfile)
|
||||
a.mtx.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = Rename(tmpfile.Name(), wfilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.mtx.Lock()
|
||||
a.dirty = false
|
||||
a.mtx.Unlock()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteExport writes an account to a special export directory named
|
||||
// by dirName. Any previous files are overwritten.
|
||||
func (a *Account) WriteExport(dirName string) error {
|
||||
exportPath := filepath.Join(networkDir(cfg.Net()), dirName)
|
||||
if err := checkCreateDir(exportPath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
aname := a.Name()
|
||||
wfilepath := accountFilename("wallet.bin", aname, exportPath)
|
||||
txfilepath := accountFilename("tx.bin", aname, exportPath)
|
||||
utxofilepath := accountFilename("utxo.bin", aname, exportPath)
|
||||
|
||||
utxofile, err := os.Create(utxofilepath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer utxofile.Close()
|
||||
a.UtxoStore.RLock()
|
||||
_, err = a.UtxoStore.s.WriteTo(utxofile)
|
||||
_, err = a.UtxoStore.s.WriteTo(tmpfile)
|
||||
a.UtxoStore.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
txfile, err := os.Create(txfilepath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer txfile.Close()
|
||||
a.TxStore.RLock()
|
||||
_, err = a.TxStore.s.WriteTo(txfile)
|
||||
a.TxStore.RUnlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
wfile, err := os.Create(wfilepath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer wfile.Close()
|
||||
a.mtx.RLock()
|
||||
_, err = a.Wallet.WriteTo(wfile)
|
||||
a.mtx.RUnlock()
|
||||
if err != nil {
|
||||
if err = Rename(tmpfile.Name(), utxofilepath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue