Added context and waitgroups to goroutines to enable complete shutdown of network go routines when a node is deleted.

Ensures a goroutine won't be updating files preventing their deletion.
This commit is contained in:
Matthew R. Kasun
2022-02-07 22:56:03 -05:00
parent 22d9d4e88a
commit b8a0230c03
2 changed files with 34 additions and 15 deletions

View File

@@ -303,8 +303,8 @@ func WipeLocal(network string) error {
log.Println(err.Error()) log.Println(err.Error())
} }
} }
if ncutils.FileExists(home + "nm-" + network + ".conf") { if ncutils.FileExists(home + ifacename + ".conf") {
err = os.Remove(home + "nm-" + network + ".conf") err = os.Remove(home + ifacename + ".conf")
if err != nil { if err != nil {
log.Println("error removing .conf:") log.Println("error removing .conf:")
log.Println(err.Error()) log.Println(err.Error())

View File

@@ -28,6 +28,7 @@ import (
// == Message Caches == // == Message Caches ==
var keepalive = new(sync.Map) var keepalive = new(sync.Map)
var messageCache = new(sync.Map) var messageCache = new(sync.Map)
var networkcontext = new(sync.Map)
const lastNodeUpdate = "lnu" const lastNodeUpdate = "lnu"
const lastPeerUpdate = "lpu" const lastPeerUpdate = "lpu"
@@ -65,21 +66,26 @@ func read(network, which string) string {
// Daemon runs netclient daemon from command line // Daemon runs netclient daemon from command line
func Daemon() error { func Daemon() error {
ctx, cancel := context.WithCancel(context.Background())
networks, err := ncutils.GetSystemNetworks() networks, err := ncutils.GetSystemNetworks()
if err != nil { if err != nil {
cancel()
return err return err
} }
for _, network := range networks { for _, network := range networks {
ctx, cancel := context.WithCancel(context.Background())
networkcontext.Store(network, cancel)
go MessageQueue(ctx, network) go MessageQueue(ctx, network)
} }
quit := make(chan os.Signal, 1) quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGTERM, os.Interrupt) signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
<-quit <-quit
cancel() for _, network := range networks {
if cancel, ok := networkcontext.Load(network); ok {
cancel.(context.CancelFunc)()
}
}
ncutils.Log("all done") ncutils.Log("all done")
return nil return nil
} }
// SetupMQTT creates a connection to broker and return client // SetupMQTT creates a connection to broker and return client
@@ -87,7 +93,6 @@ func SetupMQTT(cfg *config.ClientConfig) mqtt.Client {
opts := mqtt.NewClientOptions() opts := mqtt.NewClientOptions()
server := getServerAddress(cfg) server := getServerAddress(cfg)
opts.AddBroker(server + ":1883") opts.AddBroker(server + ":1883")
opts.SetDefaultPublishHandler(All) opts.SetDefaultPublishHandler(All)
client := mqtt.NewClient(opts) client := mqtt.NewClient(opts)
tperiod := time.Now().Add(12 * time.Second) tperiod := time.Now().Add(12 * time.Second)
@@ -197,10 +202,18 @@ func MessageQueue(ctx context.Context, network string) {
ncutils.Log("leader not defined for network " + cfg.Node.Network) ncutils.Log("leader not defined for network " + cfg.Node.Network)
} }
defer client.Disconnect(250) defer client.Disconnect(250)
go MonitorKeepalive(ctx, client, &cfg) wg := &sync.WaitGroup{}
go Checkin(ctx, &cfg, network) wg.Add(2)
keepalivectx, keepalivecancel := context.WithCancel(context.Background())
go MonitorKeepalive(keepalivectx, wg, client, &cfg)
checkinctx, checkincancel := context.WithCancel(context.Background())
go Checkin(checkinctx, wg, &cfg, network)
<-ctx.Done() <-ctx.Done()
ncutils.Log("shutting down daemon") keepalivecancel()
checkincancel()
ncutils.Log("shutting down message queue for network " + network)
wg.Wait()
ncutils.Log("shutdown complete")
} }
// All -- mqtt message hander for all ('#') topics // All -- mqtt message hander for all ('#') topics
@@ -254,10 +267,13 @@ func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
cfg.Node = newNode cfg.Node = newNode
switch newNode.Action { switch newNode.Action {
case models.NODE_DELETE: case models.NODE_DELETE:
ncutils.Log("delete recieved") if cancel, ok := networkcontext.Load(newNode.Network); ok {
if token := client.Unsubscribe(fmt.Sprintf("update/%s/%s", newNode.Network, newNode.ID), fmt.Sprintf("peers/%s/%s", newNode.Network, newNode.ID)); token.Wait() && token.Error() != nil { ncutils.Log("cancelling message queue context for " + newNode.Network)
ncutils.PrintLog("error unsubscribing during node deletion", 1) cancel.(context.CancelFunc)()
} else {
ncutils.Log("failed to kill go routines for network " + newNode.Network)
} }
ncutils.Log("deleting configuration files")
if err := WipeLocal(cfg.Network); err != nil { if err := WipeLocal(cfg.Network); err != nil {
ncutils.PrintLog("error deleting local instance: "+err.Error(), 1) ncutils.PrintLog("error deleting local instance: "+err.Error(), 1)
ncutils.PrintLog("Please perform manual clean up", 1) ncutils.PrintLog("Please perform manual clean up", 1)
@@ -298,7 +314,7 @@ func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
} }
if ifaceDelta { if ifaceDelta {
ncutils.Log("applying WG conf to " + file) ncutils.Log("applying WG conf to " + file)
err = wireguard.ApplyWGQuickConf(file, cfg.Node.Interface) err = wireguard.ApplyConf(&cfg.Node, cfg.Node.Interface, file)
if err != nil { if err != nil {
ncutils.Log("error restarting wg after node update " + err.Error()) ncutils.Log("error restarting wg after node update " + err.Error())
return return
@@ -370,10 +386,12 @@ func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
} }
// MonitorKeepalive - checks time last server keepalive received. If more than 3+ minutes, notify and resubscribe // MonitorKeepalive - checks time last server keepalive received. If more than 3+ minutes, notify and resubscribe
func MonitorKeepalive(ctx context.Context, client mqtt.Client, cfg *config.ClientConfig) { func MonitorKeepalive(ctx context.Context, wg *sync.WaitGroup, client mqtt.Client, cfg *config.ClientConfig) {
defer wg.Done()
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
ncutils.Log("cancel recieved, monitor keepalive exiting")
return return
case <-time.After(time.Second * 150): case <-time.After(time.Second * 150):
var keepalivetime time.Time var keepalivetime time.Time
@@ -467,7 +485,8 @@ func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) error {
// Checkin -- go routine that checks for public or local ip changes, publishes changes // Checkin -- go routine that checks for public or local ip changes, publishes changes
// if there are no updates, simply "pings" the server as a checkin // if there are no updates, simply "pings" the server as a checkin
func Checkin(ctx context.Context, cfg *config.ClientConfig, network string) { func Checkin(ctx context.Context, wg *sync.WaitGroup, cfg *config.ClientConfig, network string) {
defer wg.Done()
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():