Skip to content

Commit

Permalink
Modernize the RPC server.
Browse files Browse the repository at this point in the history 
This is a rather monolithic commit that moves the old RPC server to
its own package (rpc/legacyrpc), introduces a new RPC server using
gRPC (rpc/rpcserver), and provides the ability to defer wallet loading
until request at a later time by an RPC (--noinitialload).

The legacy RPC server remains the default for now while the new gRPC
server is not enabled by default.  Enabling the new server requires
setting a listen address (--experimenalrpclisten).  This experimental
flag is used to effectively feature gate the server until it is ready
to use as a default.  Both RPC servers can be run at the same time,
but require binding to different listen addresses.

In theory, with the legacy RPC server now living in its own package it
should become much easier to unit test the handlers.  This will be
useful for any future changes to the package, as compatibility with
Core's wallet is still desired.

Type safety has also been improved in the legacy RPC server.  Multiple
handler types are now used for methods that do and do not require the
RPC client as a dependency.  This can statically help prevent nil
pointer dereferences, and was very useful for catching bugs during
refactoring.

To synchronize the wallet loading process between the main package
(the default) and through the gRPC WalletLoader service (with the
--noinitialload option), as well as increasing the loose coupling of
packages, a new wallet.Loader type has been added.  All creating and
loading of existing wallets is done through a single Loader instance,
and callbacks can be attached to the instance to run after the wallet
has been opened.  This is how the legacy RPC server is associated with
a loaded wallet, even after the wallet is loaded by a gRPC method in a
completely unrelated package.

Documentation for the new RPC server has been added to the
rpc/documentation directory.  The documentation includes a
specification for the new RPC API, addresses how to make changes to
the server implementation, and provides short example clients in
several different languages.

Some of the new RPC methods are not implementated exactly as described
by the specification.  These are considered bugs with the
implementation, not the spec.  Known bugs are commented as such.
  • Loading branch information
@jrick
jrick committed Jan 29, 2016
1 parent 6af96bf commit 497ffc1
Show file tree
Hide file tree
Showing 39 changed files with 9,871 additions and 3,945 deletions.
219 changes: 152 additions & 67 deletions btcwallet.go
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2013, 2014 The btcsuite developers
* Copyright (c) 2013-2015 The btcsuite developers
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
Expand All @@ -23,8 +23,12 @@ import (
_ "net/http/pprof"
"os"
"runtime"
"sync"

"github.com/btcsuite/btcwallet/chain"
"github.com/btcsuite/btcwallet/rpc/legacyrpc"
"github.com/btcsuite/btcwallet/wallet"
"github.com/btcsuite/btcwallet/walletdb"
)

var (
Expand Down Expand Up @@ -67,88 +71,169 @@ func walletMain() error {
}()
}

// Load the wallet database. It must have been created with the
// --create option already or this will return an appropriate error.
wallet, db, err := openWallet()
if err != nil {
log.Errorf("%v", err)
return err
}
defer db.Close()
dbDir := networkDir(cfg.DataDir, activeNet.Params)
loader := wallet.NewLoader(activeNet.Params, dbDir)

// Create and start HTTP server to serve wallet client connections.
// This will be updated with the wallet and chain server RPC client
// created below after each is created.
server, err := newRPCServer(cfg.SvrListeners, cfg.RPCMaxClients,
cfg.RPCMaxWebsockets)
rpcs, legacyRPCServer, err := startRPCServers(loader)
if err != nil {
log.Errorf("Unable to create HTTP server: %v", err)
log.Errorf("Unable to create RPC servers: %v", err)
return err
}
server.Start()
server.SetWallet(wallet)

// Shutdown the server if an interrupt signal is received.
addInterruptHandler(server.Stop)

go func() {
for {
// Read CA certs and create the RPC client.
var certs []byte
if !cfg.DisableClientTLS {
certs, err = ioutil.ReadFile(cfg.CAFile)
if err != nil {
log.Warnf("Cannot open CA file: %v", err)
// If there's an error reading the CA file, continue
// with nil certs and without the client connection
certs = nil
}
} else {
log.Info("Client TLS is disabled")
}
rpcc, err := chain.NewClient(activeNet.Params, cfg.RPCConnect,
cfg.BtcdUsername, cfg.BtcdPassword, certs, cfg.DisableClientTLS)

// Create and start chain RPC client so it's ready to connect to
// the wallet when loaded later.
if !cfg.NoInitialLoad {
go rpcClientConnectLoop(legacyRPCServer, loader)
}

var closeDB func() error
defer func() {
if closeDB != nil {
err := closeDB()
if err != nil {
log.Errorf("Cannot create chain server RPC client: %v", err)
return
log.Errorf("Unable to close wallet database: %v", err)
}
err = rpcc.Start()
if err != nil {
log.Warnf("Connection to Bitcoin RPC chain server " +
"unsuccessful -- available RPC methods will be limited")
}
}()
loader.RunAfterLoad(func(w *wallet.Wallet, db walletdb.DB) {
startWalletRPCServices(w, rpcs, legacyRPCServer)
closeDB = db.Close
})

if !cfg.NoInitialLoad {
// Load the wallet database. It must have been created already
// or this will return an appropriate error.
_, err = loader.OpenExistingWallet([]byte(cfg.WalletPass), true)
if err != nil {
log.Error(err)
return err
}
}

// Shutdown the server(s) when interrupt signal is received.
if rpcs != nil {
addInterruptHandler(func() {
// TODO: Does this need to wait for the grpc server to
// finish up any requests?
log.Warn("Stopping RPC server...")
rpcs.Stop()
log.Info("RPC server shutdown")
})
}
if legacyRPCServer != nil {
go func() {
<-legacyRPCServer.RequestProcessShutdown()
simulateInterrupt()
}()
addInterruptHandler(func() {
log.Warn("Stopping legacy RPC server...")
legacyRPCServer.Stop()
log.Info("Legacy RPC server shutdown")
})
}

<-interruptHandlersDone
log.Info("Shutdown complete")
return nil
}

// rpcClientConnectLoop continuously attempts a connection to the consensus RPC
// server. When a connection is established, the client is used to sync the
// loaded wallet, either immediately or when loaded at a later time.
//
// The legacy RPC is optional. If set, the connected RPC client will be
// associated with the server for RPC passthrough and to enable additional
// methods.
func rpcClientConnectLoop(legacyRPCServer *legacyrpc.Server, loader *wallet.Loader) {
certs := readCAFile()

for {
chainClient, err := startChainRPC(certs)
if err != nil {
log.Errorf("Unable to open connection to consensus RPC server: %v", err)
continue
}

// Rather than inlining this logic directly into the loader
// callback, a function variable is used to avoid running any of
// this after the client disconnects by setting it to nil. This
// prevents the callback from associating a wallet loaded at a
// later time with a client that has already disconnected. A
// mutex is used to make this concurrent safe.
associateRPCClient := func(w *wallet.Wallet) {
w.SynchronizeRPC(chainClient)
if legacyRPCServer != nil {
legacyRPCServer.SetChainServer(chainClient)
}
// Even if Start errored, we still add the server disconnected.
// All client methods will then error, so it's obvious to a
// client that the there was a connection problem.
server.SetChainServer(rpcc)

// Start wallet goroutines and handle RPC client notifications
// if the server is not shutting down.
select {
case <-server.quit:
return
default:
wallet.Start(rpcc)
}
mu := new(sync.Mutex)
loader.RunAfterLoad(func(w *wallet.Wallet, db walletdb.DB) {
mu.Lock()
associate := associateRPCClient
mu.Unlock()
if associate != nil {
associate(w)
}
})

// Block goroutine until the client is finished.
rpcc.WaitForShutdown()
chainClient.WaitForShutdown()

wallet.SetChainSynced(false)
wallet.Stop()
mu.Lock()
associateRPCClient = nil
mu.Unlock()

// Reconnect only if the server is not shutting down.
select {
case <-server.quit:
loadedWallet, ok := loader.LoadedWallet()
if ok {
// Do not attempt a reconnect when the wallet was
// explicitly stopped.
if loadedWallet.ShuttingDown() {
return
default:
}

loadedWallet.SetChainSynced(false)

// TODO: Rework the wallet so changing the RPC client
// does not require stopping and restarting everything.
loadedWallet.Stop()
loadedWallet.WaitForShutdown()
loadedWallet.Start()
}
}()
}
}

// Wait for the server to shutdown either due to a stop RPC request
// or an interrupt.
server.WaitForShutdown()
log.Info("Shutdown complete")
return nil
func readCAFile() []byte {
// Read certificate file if TLS is not disabled.
var certs []byte
if !cfg.DisableClientTLS {
var err error
certs, err = ioutil.ReadFile(cfg.CAFile)
if err != nil {
log.Warnf("Cannot open CA file: %v", err)
// If there's an error reading the CA file, continue
// with nil certs and without the client connection.
certs = nil
}
} else {
log.Info("Chain server RPC TLS is disabled")
}

return certs
}

// startChainRPC opens a RPC client connection to a btcd server for blockchain
// services. This function uses the RPC options from the global config and
// there is no recovery in case the server is not available or if there is an
// authentication error. Instead, all requests to the client will simply error.
func startChainRPC(certs []byte) (*chain.RPCClient, error) {
log.Infof("Attempting RPC client connection to %v", cfg.RPCConnect)
rpcc, err := chain.NewRPCClient(activeNet.Params, cfg.RPCConnect,
cfg.BtcdUsername, cfg.BtcdPassword, certs, cfg.DisableClientTLS, 0)
if err != nil {
return nil, err
}
err = rpcc.Start()
return rpcc, err
}
Loading

0 comments on commit 497ffc1

Please sign in to comment.