use atomic for lock-free

This commit is contained in:
aliamerj
2025-09-17 12:45:35 +03:00
parent 889c3c8f8d
commit 3027029f7b
+20 -29
View File
@@ -14,6 +14,7 @@ import (
"sort" "sort"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/hashicorp/go-multierror" "github.com/hashicorp/go-multierror"
@@ -199,8 +200,8 @@ type Engine struct {
srWatcher *guard.SRWatcher srWatcher *guard.SRWatcher
// Sync response persistence // Sync response persistence
persistSyncResponse bool persistSyncResponse atomic.Bool
latestSyncResponse *mgmProto.SyncResponse latestSyncResponse atomic.Pointer[mgmProto.SyncResponse]
connSemaphore *semaphoregroup.SemaphoreGroup connSemaphore *semaphoregroup.SemaphoreGroup
flowManager nftypes.FlowManager flowManager nftypes.FlowManager
@@ -661,10 +662,8 @@ func (e *Engine) removePeer(peerKey string) error {
return nil return nil
} }
// 1. xyzzz
func (e *Engine) handleSync(update *mgmProto.SyncResponse) error { func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
e.syncMsgMux.Lock()
defer e.syncMsgMux.Unlock()
if update.GetNetbirdConfig() != nil { if update.GetNetbirdConfig() != nil {
wCfg := update.GetNetbirdConfig() wCfg := update.GetNetbirdConfig()
err := e.updateTURNs(wCfg.GetTurns()) err := e.updateTURNs(wCfg.GetTurns())
@@ -703,10 +702,11 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
if nm == nil { if nm == nil {
return nil return nil
} }
persistSyncResponse := e.persistSyncResponse.Load()
// Store sync response if persistence is enabled // Store sync response if persistence is enabled
if e.persistSyncResponse { if persistSyncResponse {
e.latestSyncResponse = update e.latestSyncResponse.Store(update)
log.Debugf("sync response persisted with serial %d", nm.GetSerial()) log.Debugf("sync response persisted with serial %d", nm.GetSerial())
} }
@@ -927,16 +927,9 @@ func (e *Engine) receiveJobEvents() {
} }
func (e *Engine) handleBundle(params *mgmProto.BundleParameters) (*mgmProto.JobResponse_Bundle, error) { func (e *Engine) handleBundle(params *mgmProto.BundleParameters) (*mgmProto.JobResponse_Bundle, error) {
// Access sync response directly without calling exported method to avoid deadlock syncResponse, err := e.GetLatestSyncResponse()
// This is safe because we're in a goroutine that doesn't hold the syncMsgMux if err != nil {
var syncResponse *mgmProto.SyncResponse return nil, fmt.Errorf("get latest sync response: %w", err)
if e.persistSyncResponse && e.latestSyncResponse != nil {
log.Debugf("Retrieving latest sync response with size %d bytes", proto.Size(e.latestSyncResponse))
if sr, ok := proto.Clone(e.latestSyncResponse).(*mgmProto.SyncResponse); ok {
syncResponse = sr
} else {
return nil, fmt.Errorf("failed to clone sync response")
}
} }
if syncResponse == nil { if syncResponse == nil {
@@ -1823,36 +1816,34 @@ func (e *Engine) stopDNSServer() {
// SetSyncResponsePersistence enables or disables sync response persistence // SetSyncResponsePersistence enables or disables sync response persistence
func (e *Engine) SetSyncResponsePersistence(enabled bool) { func (e *Engine) SetSyncResponsePersistence(enabled bool) {
e.syncMsgMux.Lock() persistSyncResponse := e.persistSyncResponse.Load()
defer e.syncMsgMux.Unlock() if enabled == persistSyncResponse {
if enabled == e.persistSyncResponse {
return return
} }
e.persistSyncResponse = enabled e.persistSyncResponse.Store(enabled)
log.Debugf("Sync response persistence is set to %t", enabled) log.Debugf("Sync response persistence is set to %t", enabled)
if !enabled { if !enabled {
e.latestSyncResponse = nil e.latestSyncResponse.Store(nil)
} }
} }
// GetLatestSyncResponse returns the stored sync response if persistence is enabled // GetLatestSyncResponse returns the stored sync response if persistence is enabled
func (e *Engine) GetLatestSyncResponse() (*mgmProto.SyncResponse, error) { func (e *Engine) GetLatestSyncResponse() (*mgmProto.SyncResponse, error) {
e.syncMsgMux.Lock() persistSyncResponse := e.persistSyncResponse.Load()
defer e.syncMsgMux.Unlock()
if !e.persistSyncResponse { if !persistSyncResponse {
return nil, errors.New("sync response persistence is disabled") return nil, errors.New("sync response persistence is disabled")
} }
if e.latestSyncResponse == nil { latestSyncResponse := e.latestSyncResponse.Load()
if latestSyncResponse == nil {
//nolint:nilnil //nolint:nilnil
return nil, nil return nil, nil
} }
log.Debugf("Retrieving latest sync response with size %d bytes", proto.Size(e.latestSyncResponse)) log.Debugf("Retrieving latest sync response with size %d bytes", proto.Size(latestSyncResponse))
sr, ok := proto.Clone(e.latestSyncResponse).(*mgmProto.SyncResponse) sr, ok := proto.Clone(latestSyncResponse).(*mgmProto.SyncResponse)
if !ok { if !ok {
return nil, fmt.Errorf("failed to clone sync response") return nil, fmt.Errorf("failed to clone sync response")
} }