From 9c200097f818627a01c8c7aed71dbfd6265df610 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 21 Jun 2022 15:11:50 -0400 Subject: [PATCH] Fix various races detected by the race detector --- controller/sync_strats/sync_instant.go | 19 ++++++----- router/handler_edge_ctrl/apiSessionAdded.go | 9 +++-- router/xgress_edge/certchecker_test.go | 6 ++-- tunnel/dns/server.go | 37 +++++++++++++-------- 4 files changed, 40 insertions(+), 31 deletions(-) diff --git a/controller/sync_strats/sync_instant.go b/controller/sync_strats/sync_instant.go index d8fd709c2..0c19dd698 100644 --- a/controller/sync_strats/sync_instant.go +++ b/controller/sync_strats/sync_instant.go @@ -19,7 +19,6 @@ package sync_strats import ( "encoding/json" "fmt" - "google.golang.org/protobuf/proto" "github.com/lucsky/cuid" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" @@ -30,9 +29,11 @@ import ( "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/build" "github.com/openziti/fabric/controller/network" - "github.com/openziti/storage/ast" + "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/foundation/util/debugz" + "github.com/openziti/storage/ast" "go.etcd.io/bbolt" + "google.golang.org/protobuf/proto" "strings" "sync" "time" @@ -87,7 +88,8 @@ type InstantStrategy struct { routerConnectedQueue chan *RouterSender receivedClientHelloQueue chan *RouterSender - stop chan struct{} + stopNotify chan struct{} + stopped concurrenz.AtomicBoolean } func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *InstantStrategy { @@ -124,7 +126,7 @@ func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *Instant routerConnectedQueue: make(chan *RouterSender, options.MaxQueuedRouterConnects), receivedClientHelloQueue: make(chan *RouterSender, options.MaxQueuedClientHellos), - stop: make(chan struct{}, 0), + stopNotify: make(chan struct{}, 0), } strategy.helloHandler = handler_edge_ctrl.NewHelloHandler(ae, strategy.ReceiveClientHello) @@ -150,9 +152,8 @@ func (strategy *InstantStrategy) Type() env.RouterSyncStrategyType { } func (strategy *InstantStrategy) Stop() { - if strategy.stop != nil { - close(strategy.stop) - strategy.stop = nil + if strategy.stopped.CompareAndSwap(false, true) { + close(strategy.stopNotify) } } @@ -267,7 +268,7 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) { func (strategy *InstantStrategy) startHandleRouterConnectWorker() { for { select { - case <-strategy.stop: + case <-strategy.stopNotify: return case rtx := <-strategy.routerConnectedQueue: func() { @@ -287,7 +288,7 @@ func (strategy *InstantStrategy) startHandleRouterConnectWorker() { func (strategy *InstantStrategy) startSynchronizeWorker() { for { select { - case <-strategy.stop: + case <-strategy.stopNotify: return case rtx := <-strategy.receivedClientHelloQueue: func() { diff --git a/router/handler_edge_ctrl/apiSessionAdded.go b/router/handler_edge_ctrl/apiSessionAdded.go index d477ba04c..a3313da90 100644 --- a/router/handler_edge_ctrl/apiSessionAdded.go +++ b/router/handler_edge_ctrl/apiSessionAdded.go @@ -75,8 +75,11 @@ func (h *apiSessionAddedHandler) HandleReceive(msg *channel.Message, _ channel.C go func() { req := &edge_ctrl_pb.ApiSessionAdded{} if err := proto.Unmarshal(msg.Body, req); err == nil { - if req.IsFullState { + for _, session := range req.ApiSessions { + h.sm.AddApiSession(session) + } + if req.IsFullState { reqWithState := &apiSessionAddedWithState{ ApiSessionAdded: req, } @@ -98,10 +101,6 @@ func (h *apiSessionAddedHandler) HandleReceive(msg *channel.Message, _ channel.C } h.reqChan <- reqWithState } - - for _, session := range req.ApiSessions { - h.sm.AddApiSession(session) - } } else { pfxlog.Logger().Panic("could not convert message as api session added") } diff --git a/router/xgress_edge/certchecker_test.go b/router/xgress_edge/certchecker_test.go index fbc14ae98..e6d123b54 100644 --- a/router/xgress_edge/certchecker_test.go +++ b/router/xgress_edge/certchecker_test.go @@ -254,11 +254,11 @@ func Test_CertExpirationChecker(t *testing.T) { certChecker := newCertChecker() certChecker.timeoutDuration = 10 * time.Millisecond - invoked := false + var invoked concurrenz.AtomicBoolean extender := &stubExtender{ done: func() error { - invoked = true + invoked.Set(true) certChecker.id.Cert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0) certChecker.id.ServerCert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0) return errors.New("test") @@ -275,7 +275,7 @@ func Test_CertExpirationChecker(t *testing.T) { time.Sleep(200 * time.Millisecond) - req.True(invoked) + req.True(invoked.Get()) certChecker.closeNotify <- struct{}{} }) diff --git a/tunnel/dns/server.go b/tunnel/dns/server.go index 6df044c69..60751b0f4 100644 --- a/tunnel/dns/server.go +++ b/tunnel/dns/server.go @@ -84,6 +84,17 @@ func NewDnsServer(addr string) Resolver { Net: "udp", } + names := make(map[string]net.IP) + r := &resolver{ + server: s, + names: names, + ips: make(map[string]string), + namesMtx: sync.Mutex{}, + domains: make(map[string]*domainEntry), + domainsMtx: sync.Mutex{}, + } + s.Handler = r + errChan := make(chan error) go func() { errChan <- s.ListenAndServe() @@ -100,17 +111,6 @@ func NewDnsServer(addr string) Resolver { log.Infof("dns server running at %s", s.Addr) } - names := make(map[string]net.IP) - r := &resolver{ - server: s, - names: names, - ips: make(map[string]string), - namesMtx: sync.Mutex{}, - domains: make(map[string]*domainEntry), - domainsMtx: sync.Mutex{}, - } - s.Handler = r - const resolverConfigHelp = "ziti-tunnel runs an internal DNS server which must be first in the host's\n" + "resolver configuration. On systems that use NetManager/dhclient, this can\n" + "be achieved by adding the following to /etc/dhcp/dhclient.conf:\n" + @@ -150,13 +150,22 @@ func (r *resolver) testSystemResolver() error { return nil } -func (r *resolver) getAddress(name string) (net.IP, error) { +func (r *resolver) getHostnameIp(name string) (net.IP, bool) { + r.namesMtx.Lock() + defer r.namesMtx.Unlock() canonical := strings.ToLower(name) - a, ok := r.names[canonical] + ip, found := r.names[canonical] + return ip, found +} + +func (r *resolver) getAddress(name string) (net.IP, error) { + a, ok := r.getHostnameIp(name) if ok { return a, nil } + canonical := strings.ToLower(name) + r.domainsMtx.Lock() defer r.domainsMtx.Unlock() for { @@ -174,7 +183,7 @@ func (r *resolver) getAddress(name string) (net.IP, error) { return nil, err } log.Debugf("assigned %v => %v", name, ip) - r.AddHostname(name[:len(name)-1], ip) + _ = r.AddHostname(name[:len(name)-1], ip) // this resolver impl never returns an error return ip, err } }