From 6d399a6525511e3f7e4a12e5fa842d09795cbd0a Mon Sep 17 00:00:00 2001
From: Angelo Capossele <angelocapossele@gmail.com>
Date: Tue, 23 Jun 2020 19:23:07 +0100
Subject: [PATCH] Fix graceful shutdown (#527)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
* 🐛 Fix grpc server race condition
* 🐛 Fix graceful shutdown
* 👌 Add nil check
* ♻️ Refactor FPC gRPC server
* stop worker on server issues
Co-authored-by: Wolfgang Welz <welzwo@gmail.com>
---
dapps/valuetransfers/fpc.go | 25 ++++++++++++++++++-------
packages/vote/net/server.go | 3 +--
plugins/analysis/client/connector.go | 2 +-
3 files changed, 20 insertions(+), 10 deletions(-)
diff --git a/dapps/valuetransfers/fpc.go b/dapps/valuetransfers/fpc.go
index 850202eb..3bf6bd9c 100644
--- a/dapps/valuetransfers/fpc.go
+++ b/dapps/valuetransfers/fpc.go
@@ -101,7 +101,10 @@ func configureFPC() {
}
func runFPC() {
- if err := daemon.BackgroundWorker("FPCVoterServer", func(shutdownSignal <-chan struct{}) {
+ const ServerWorkerName = "FPCVoterServer"
+ if err := daemon.BackgroundWorker(ServerWorkerName, func(shutdownSignal <-chan struct{}) {
+ stopped := make(chan struct{})
+ bindAddr := config.Node.GetString(CfgFPCBindAddress)
voterServer = votenet.New(Voter(), func(id string) vote.Opinion {
branchID, err := branchmanager.BranchIDFromBase58(id)
if err != nil {
@@ -123,21 +126,29 @@ func runFPC() {
}
return vote.Like
- }, config.Node.GetString(CfgFPCBindAddress),
+ }, bindAddr,
metrics.Events().FPCInboundBytes,
metrics.Events().FPCOutboundBytes,
- metrics.Events().QueryReceived)
+ metrics.Events().QueryReceived,
+ )
go func() {
+ log.Infof("%s started, bind-address=%s", ServerWorkerName, bindAddr)
if err := voterServer.Run(); err != nil {
- log.Error(err)
+ log.Errorf("Error serving: %s", err)
}
+ close(stopped)
}()
- log.Infof("Started vote server on %s", config.Node.GetString(CfgFPCBindAddress))
- <-shutdownSignal
+ // stop if we are shutting down or the server could not be started
+ select {
+ case <-shutdownSignal:
+ case <-stopped:
+ }
+
+ log.Infof("Stopping %s ...", ServerWorkerName)
voterServer.Shutdown()
- log.Info("Stopped vote server")
+ log.Infof("Stopping %s ... done", ServerWorkerName)
}, shutdown.PriorityFPC); err != nil {
log.Panicf("Failed to start as daemon: %s", err)
}
diff --git a/packages/vote/net/server.go b/packages/vote/net/server.go
index d6207a73..97209c50 100644
--- a/packages/vote/net/server.go
+++ b/packages/vote/net/server.go
@@ -21,6 +21,7 @@ func New(voter vote.Voter, opnRetriever OpinionRetriever, bindAddr string, netRx
voter: voter,
opnRetriever: opnRetriever,
bindAddr: bindAddr,
+ grpcServer: grpc.NewServer(),
netRxEvent: netRxEvent,
netTxEvent: netTxEvent,
queryReceivedEvent: queryReceivedEvent,
@@ -71,9 +72,7 @@ func (vs *VoterServer) Run() error {
return err
}
- vs.grpcServer = grpc.NewServer()
RegisterVoterQueryServer(vs.grpcServer, vs)
-
return vs.grpcServer.Serve(listener)
}
diff --git a/plugins/analysis/client/connector.go b/plugins/analysis/client/connector.go
index cd9da66a..269b7647 100644
--- a/plugins/analysis/client/connector.go
+++ b/plugins/analysis/client/connector.go
@@ -93,7 +93,7 @@ func (c *Connector) dial() {
return
default:
c.conn = nil
- conn, err := net.Dial(c.network, c.address)
+ conn, err := net.DialTimeout(c.network, c.address, 5*time.Second)
if err != nil {
go c.scheduleRedial()
return
--
GitLab