loopd: refactor into Start/Stop methods

cmd/loopd/main.go

@@ -10,6 +10,6 @@ func main() {
 	cfg := loopd.RPCConfig{}
 	err := loopd.Start(cfg)
 	if err != nil {
-		fmt.Println(err)
+		fmt.Printf("loopd exited with an error: %v\n", err)
 	}
 }

loopd/daemon.go

@@ -6,11 +6,8 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"os"
-	"os/signal"
-	"runtime/pprof"
 	"sync"
-	"time"
+	"sync/atomic"
 
 	proxy "github.com/grpc-ecosystem/grpc-gateway/runtime"
 	"github.com/lightninglabs/loop"
@@ -24,6 +21,10 @@ var (
 	// maxMsgRecvSize is the largest message our REST proxy will receive. We
 	// set this to 200MiB atm.
 	maxMsgRecvSize = grpc.MaxCallRecvMsgSize(1 * 1024 * 1024 * 200)
+
+	// errOnlyStartOnce is the error that is returned if the daemon is
+	// started more than once.
+	errOnlyStartOnce = fmt.Errorf("daemon can only be started once")
 )
 
 // listenerCfg holds closures used to retrieve listeners for the gRPC services.
@@ -40,36 +41,208 @@ type listenerCfg struct {
 
 // Daemon is the struct that holds one instance of the loop client daemon.
 type Daemon struct {
+	// To be used atomically. Declared first to optimize struct alignment.
+	started int32
+
 	// swapClientServer is the embedded RPC server that satisfies the client
 	// RPC interface. We embed this struct so the Daemon itself can be
 	// registered to an existing grpc.Server to run as a subserver in the
 	// same process.
 	swapClientServer
 
-	cfg         *Config
-	listenerCfg *listenerCfg
+	// ErrChan is an error channel that users of the Daemon struct must use
+	// to detect runtime errors and also whether a shutdown is fully
+	// completed.
+	ErrChan chan error
 
-	lnd *lndclient.GrpcLndServices
+	cfg             *Config
+	listenerCfg     *listenerCfg
+	internalErrChan chan error
+
+	lnd           *lndclient.GrpcLndServices
+	clientCleanup func()
+
+	wg       sync.WaitGroup
+	quit     chan struct{}
+	stopOnce sync.Once
+
+	mainCtx       context.Context
+	mainCtxCancel func()
+
+	grpcServer    *grpc.Server
+	grpcListener  net.Listener
+	restServer    *http.Server
+	restListener  net.Listener
+	restCtxCancel func()
 }
 
 // New creates a new instance of the loop client daemon.
 func New(config *Config, lisCfg *listenerCfg) *Daemon {
 	return &Daemon{
+		// We send exactly one error on this channel if something goes
+		// wrong at runtime. Or a nil value if the shutdown was
+		// successful. But in case nobody's listening, we don't want to
+		// block on it so we buffer it.
+		ErrChan: make(chan error, 1),
+
+		quit:        make(chan struct{}),
 		cfg:         config,
 		listenerCfg: lisCfg,
+
+		// We have 3 goroutines that could potentially send an error.
+		// We react on the first error but in case more than one exits
+		// with an error we don't want them to block.
+		internalErrChan: make(chan error, 3),
 	}
 }
 
-// Run runs loopd in daemon mode. It will listen for grpc connections,
+// Start starts loopd in daemon mode. It will listen for grpc connections,
 // execute commands and pass back swap status information.
-func (d *Daemon) Run() error {
+func (d *Daemon) Start() error {
+	// There should be no reason to start the daemon twice. Therefore return
+	// an error if that's tried.
+	if atomic.AddInt32(&d.started, 1) != 1 {
+		return errOnlyStartOnce
+	}
+
 	var err error
 	d.lnd, err = d.listenerCfg.getLnd(d.cfg.Network, d.cfg.Lnd)
 	if err != nil {
 		return err
 	}
-	defer d.lnd.Close()
 
+	// With lnd connected, initialize everything else, such as the swap
+	// server client, the swap client RPC server instance and our main swap
+	// and error handlers. If this fails, then nothing has been started yet
+	// and we can just return the error.
+	err = d.initialize()
+	if err != nil {
+		return err
+	}
+
+	// If we get here, we already have started several goroutines. So if
+	// anything goes wrong now, we need to cleanly shut down again.
+	startErr := d.startWebServers()
+	if startErr != nil {
+		log.Errorf("Error while starting daemon: %v", err)
+		d.Stop()
+		stopErr := <-d.ErrChan
+		if stopErr != nil {
+			log.Errorf("Error while stopping daemon: %v", stopErr)
+		}
+		return startErr
+	}
+
+	return nil
+}
+
+// startWebServers starts the gRPC and REST servers in goroutines.
+func (d *Daemon) startWebServers() error {
+	var err error
+
+	// With our client created, let's now finish setting up and start our
+	// RPC server.
+	serverOpts := []grpc.ServerOption{}
+	d.grpcServer = grpc.NewServer(serverOpts...)
+	looprpc.RegisterSwapClientServer(d.grpcServer, d)
+
+	// Next, start the gRPC server listening for HTTP/2 connections.
+	log.Infof("Starting gRPC listener")
+	d.grpcListener, err = d.listenerCfg.grpcListener()
+	if err != nil {
+		return fmt.Errorf("RPC server unable to listen on %s: %v",
+			d.cfg.RPCListen, err)
+
+	}
+
+	// The default JSON marshaler of the REST proxy only sets OrigName to
+	// true, which instructs it to use the same field names as specified in
+	// the proto file and not switch to camel case. What we also want is
+	// that the marshaler prints all values, even if they are falsey.
+	customMarshalerOption := proxy.WithMarshalerOption(
+		proxy.MIMEWildcard, &proxy.JSONPb{
+			OrigName:     true,
+			EmitDefaults: true,
+		},
+	)
+
+	// We'll also create and start an accompanying proxy to serve clients
+	// through REST.
+	ctx, cancel := context.WithCancel(context.Background())
+	d.restCtxCancel = cancel
+	mux := proxy.NewServeMux(customMarshalerOption)
+	var restHandler http.Handler = mux
+	if d.cfg.CORSOrigin != "" {
+		restHandler = allowCORS(restHandler, d.cfg.CORSOrigin)
+	}
+	proxyOpts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithDefaultCallOptions(maxMsgRecvSize),
+	}
+	err = looprpc.RegisterSwapClientHandlerFromEndpoint(
+		ctx, mux, d.cfg.RPCListen, proxyOpts,
+	)
+	if err != nil {
+		return err
+	}
+
+	d.restListener, err = d.listenerCfg.restListener()
+	if err != nil {
+		return fmt.Errorf("REST proxy unable to listen on %s: %v",
+			d.cfg.RESTListen, err)
+	}
+
+	// A nil listener indicates REST is disabled.
+	if d.restListener != nil {
+		log.Infof("Starting REST proxy listener")
+
+		d.restServer = &http.Server{Handler: restHandler}
+
+		d.wg.Add(1)
+		go func() {
+			defer d.wg.Done()
+
+			log.Infof("REST proxy listening on %s",
+				d.restListener.Addr())
+			err := d.restServer.Serve(d.restListener)
+			// ErrServerClosed is always returned when the proxy is
+			// shut down, so don't log it.
+			if err != nil && err != http.ErrServerClosed {
+				// Notify the main error handler goroutine that
+				// we exited unexpectedly here. We don't have to
+				// worry about blocking as the internal error
+				// channel is sufficiently buffered.
+				d.internalErrChan <- err
+			}
+		}()
+	} else {
+		log.Infof("REST proxy disabled")
+	}
+
+	// Start the grpc server.
+	d.wg.Add(1)
+	go func() {
+		defer d.wg.Done()
+
+		log.Infof("RPC server listening on %s", d.grpcListener.Addr())
+		err = d.grpcServer.Serve(d.grpcListener)
+		if err != nil && err != grpc.ErrServerStopped {
+			// Notify the main error handler goroutine that
+			// we exited unexpectedly here. We don't have to
+			// worry about blocking as the internal error
+			// channel is sufficiently buffered.
+			d.internalErrChan <- err
+		}
+	}()
+
+	return nil
+}
+
+// initialize creates and initializes an instance of the swap server client,
+// the swap client RPC server instance and our main swap and error handlers. If
+// this method fails with an error then no goroutine was started yet and no
+// cleanup is necessary. If it succeeds, then goroutines have been spawned.
+func (d *Daemon) initialize() error {
 	// If no swap server is specified, use the default addresses for mainnet
 	// and testnet.
 	if d.cfg.SwapServer == "" {
@@ -88,173 +261,148 @@ func (d *Daemon) Run() error {
 	log.Infof("Swap server address: %v", d.cfg.SwapServer)
 
 	// Create an instance of the loop client library.
-	swapClient, cleanup, err := getClient(d.cfg, &d.lnd.LndServices)
+	swapclient, clientCleanup, err := getClient(d.cfg, &d.lnd.LndServices)
 	if err != nil {
 		return err
 	}
-	defer cleanup()
+	d.clientCleanup = clientCleanup
 
-	// Retrieve all currently existing swaps from the database.
-	swapsList, err := swapClient.FetchSwaps()
-	if err != nil {
-		return err
-	}
+	// Both the client RPC server and and the swap server client should
+	// stop on main context cancel. So we create it early and pass it down.
+	d.mainCtx, d.mainCtxCancel = context.WithCancel(context.Background())
 
-	swaps := make(map[lntypes.Hash]loop.SwapInfo)
-	for _, s := range swapsList {
-		swaps[s.SwapHash] = *s
-	}
-
-	// Instantiate the loopd gRPC server.
+	// Now finally fully initialize the swap client RPC server instance.
 	d.swapClientServer = swapClientServer{
-		impl:        swapClient,
+		impl:        swapclient,
 		lnd:         &d.lnd.LndServices,
-		swaps:       swaps,
+		swaps:       make(map[lntypes.Hash]loop.SwapInfo),
 		subscribers: make(map[int]chan<- interface{}),
 		statusChan:  make(chan loop.SwapInfo),
+		mainCtx:     d.mainCtx,
 	}
 
-	serverOpts := []grpc.ServerOption{}
-	grpcServer := grpc.NewServer(serverOpts...)
-	looprpc.RegisterSwapClientServer(grpcServer, d)
-
-	// Next, start the gRPC server listening for HTTP/2 connections.
-	log.Infof("Starting gRPC listener")
-	grpcListener, err := d.listenerCfg.grpcListener()
-	if err != nil {
-		return fmt.Errorf("RPC server unable to listen on %s",
-			d.cfg.RPCListen)
-
-	}
-	defer grpcListener.Close()
-
-	// The default JSON marshaler of the REST proxy only sets OrigName to
-	// true, which instructs it to use the same field names as specified in
-	// the proto file and not switch to camel case. What we also want is
-	// that the marshaler prints all values, even if they are falsey.
-	customMarshalerOption := proxy.WithMarshalerOption(
-		proxy.MIMEWildcard, &proxy.JSONPb{
-			OrigName:     true,
-			EmitDefaults: true,
-		},
-	)
-
-	// We'll also create and start an accompanying proxy to serve clients
-	// through REST.
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	mux := proxy.NewServeMux(customMarshalerOption)
-	var restHandler http.Handler = mux
-	if d.cfg.CORSOrigin != "" {
-		restHandler = allowCORS(restHandler, d.cfg.CORSOrigin)
-	}
-	proxyOpts := []grpc.DialOption{
-		grpc.WithInsecure(),
-		grpc.WithDefaultCallOptions(maxMsgRecvSize),
-	}
-	err = looprpc.RegisterSwapClientHandlerFromEndpoint(
-		ctx, mux, d.cfg.RPCListen, proxyOpts,
-	)
+	// Retrieve all currently existing swaps from the database.
+	swapsList, err := d.impl.FetchSwaps()
 	if err != nil {
+		// The client is the only thing we started yet, so if we clean
+		// up its connection now, nothing else needs to be shut down at
+		// this point.
+		clientCleanup()
 		return err
 	}
 
-	restListener, err := d.listenerCfg.restListener()
-	if err != nil {
-		return fmt.Errorf("REST proxy unable to listen on %s",
-			d.cfg.RESTListen)
+	for _, s := range swapsList {
+		d.swaps[s.SwapHash] = *s
 	}
 
-	// A nil listener indicates REST is disabled.
-	if restListener != nil {
-		log.Infof("Starting REST proxy listener")
-
-		defer restListener.Close()
-		proxy := &http.Server{Handler: restHandler}
-
-		go func() {
-			err := proxy.Serve(restListener)
-			// ErrServerClosed is always returned when the proxy is
-			// shut down, so don't log it.
-			if err != nil && err != http.ErrServerClosed {
-				log.Error(err)
-			}
-		}()
-	} else {
-		log.Infof("REST proxy disabled")
-	}
-
-	mainCtx, cancel := context.WithCancel(context.Background())
-	var wg sync.WaitGroup
-
 	// Start the swap client itself.
-	wg.Add(1)
+	d.wg.Add(1)
 	go func() {
-		defer wg.Done()
+		defer d.wg.Done()
 
 		log.Infof("Starting swap client")
-		err := swapClient.Run(mainCtx, d.statusChan)
+		err := d.impl.Run(d.mainCtx, d.statusChan)
 		if err != nil {
-			log.Error(err)
+			// Notify the main error handler goroutine that
+			// we exited unexpectedly here. We don't have to
+			// worry about blocking as the internal error
+			// channel is sufficiently buffered.
+			d.internalErrChan <- err
 		}
 		log.Infof("Swap client stopped")
-
-		log.Infof("Stopping gRPC server")
-		grpcServer.Stop()
-
-		cancel()
 	}()
 
 	// Start a goroutine that broadcasts swap updates to clients.
-	wg.Add(1)
+	d.wg.Add(1)
 	go func() {
-		defer wg.Done()
+		defer d.wg.Done()
 
 		log.Infof("Waiting for updates")
-		d.processStatusUpdates(mainCtx)
+		d.processStatusUpdates(d.mainCtx)
 	}()
 
-	// Start the grpc server.
-	wg.Add(1)
+	// Last, start our internal error handler. This will return exactly one
+	// error or nil on the main error channel to inform the caller that
+	// something went wrong or that shutdown is complete. We don't add to
+	// the wait group here because this goroutine will itself wait for the
+	// stop to complete and signal its completion through the main error
+	// channel.
 	go func() {
-		defer wg.Done()
+		var runtimeErr error
 
-		log.Infof("RPC server listening on %s", grpcListener.Addr())
+		// There are only two ways this goroutine can exit. Either there
+		// is an internal error or the caller requests shutdown. In both
+		// cases we wait for the stop to complete before we signal the
+		// caller that we're done.
+		select {
+		case runtimeErr = <-d.internalErrChan:
+			log.Errorf("Runtime error in daemon, shutting down: "+
+				"%v", runtimeErr)
 
-		if restListener != nil {
-			log.Infof("REST proxy listening on %s", restListener.Addr())
+		case <-d.quit:
 		}
 
-		err = grpcServer.Serve(grpcListener)
-		if err != nil {
-			log.Error(err)
-		}
+		// We need to shutdown before sending the error on the channel,
+		// otherwise a caller might exit the process too early.
+		d.stop()
+		log.Info("Daemon exited")
+
+		// The caller expects exactly one message. So we send the error
+		// even if it's nil because we cleanly shut down.
+		d.ErrChan <- runtimeErr
 	}()
 
-	interruptChannel := make(chan os.Signal, 1)
-	signal.Notify(interruptChannel, os.Interrupt)
-
-	// Run until the users terminates loopd or an error occurred.
-	select {
-	case <-interruptChannel:
-		log.Infof("Received SIGINT (Ctrl+C).")
-
-		// TODO: Remove debug code.
-		// Debug code to dump goroutines on hanging exit.
-		go func() {
-			time.Sleep(5 * time.Second)
-			_ = pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
-		}()
-
-		cancel()
-	case <-mainCtx.Done():
-	}
-
-	wg.Wait()
-
 	return nil
 }
 
+// Stop tries to gracefully shut down the daemon. A caller needs to wait for a
+// message on the main error channel indicating that the shutdown is completed.
+func (d *Daemon) Stop() {
+	d.stopOnce.Do(func() {
+		close(d.quit)
+	})
+}
+
+// stop does the actual shutdown and blocks until all goroutines have exit.
+func (d *Daemon) stop() {
+	// First of all, we can cancel the main context that all event handlers
+	// are using. This should stop all swap activity and all event handlers
+	// should exit.
+	if d.mainCtxCancel != nil {
+		d.mainCtxCancel()
+	}
+
+	// As there is no swap activity anymore, we can forcefully shutdown the
+	// gRPC and HTTP servers now.
+	log.Infof("Stopping gRPC server")
+	if d.grpcServer != nil {
+		d.grpcServer.Stop()
+	}
+	log.Infof("Stopping REST server")
+	if d.restServer != nil {
+		// Don't return the error here, we first want to give everything
+		// else a chance to shut down cleanly.
+		err := d.restServer.Close()
+		if err != nil {
+			log.Errorf("Error stopping REST server: %v", err)
+		}
+	}
+	if d.restCtxCancel != nil {
+		d.restCtxCancel()
+	}
+
+	// Next, shut down the connections to lnd and the swap server.
+	if d.lnd != nil {
+		d.lnd.Close()
+	}
+	if d.clientCleanup != nil {
+		d.clientCleanup()
+	}
+
+	// Everything should be shutting down now, wait for completion.
+	d.wg.Wait()
+}
+
 // allowCORS wraps the given http.Handler with a function that adds the
 // Access-Control-Allow-Origin header to the response.
 func allowCORS(handler http.Handler, origin string) http.Handler {

loopd/start.go

@@ -13,6 +13,7 @@ import (
 	"github.com/lightninglabs/loop/lndclient"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/lnrpc/verrpc"
+	"github.com/lightningnetwork/lnd/signal"
 )
 
 const defaultConfigFilename = "loopd.conf"
@@ -168,8 +169,26 @@ func Start(rpcCfg RPCConfig) error {
 
 	// Execute command.
 	if parser.Active == nil {
+		signal.Intercept()
+
 		daemon := New(&config, lisCfg)
-		return daemon.Run()
+		if err := daemon.Start(); err != nil {
+			return err
+		}
+
+		select {
+		case <-signal.ShutdownChannel():
+			log.Infof("Received SIGINT (Ctrl+C).")
+			daemon.Stop()
+
+			// The above stop will return immediately. But we'll be
+			// notified on the error channel once the process is
+			// complete.
+			return <-daemon.ErrChan
+
+		case err := <-daemon.ErrChan:
+			return err
+		}
 	}
 
 	if parser.Active.Name == "view" {

loopd/swapclient_server.go

@@ -39,6 +39,7 @@ type swapClientServer struct {
 	statusChan       chan loop.SwapInfo
 	nextSubscriberID int
 	swapsLock        sync.Mutex
+	mainCtx          context.Context
 }
 
 // LoopOut initiates an loop out swap with the given parameters. The call
@@ -264,8 +265,14 @@ func (s *swapClientServer) Monitor(in *looprpc.MonitorRequest,
 			if err := send(swap); err != nil {
 				return err
 			}
+
+		// The client cancels the subscription.
 		case <-server.Context().Done():
 			return nil
+
+		// The server is shutting down.
+		case <-s.mainCtx.Done():
+			return fmt.Errorf("server is shutting down")
 		}
 	}
 }