1
0
mirror of https://github.com/AvengeMedia/DankMaterialShell.git synced 2026-08-01 19:18:28 -04:00

cups: fix sub/unsub race

port 1.5

(cherry picked from commit bab2078dfd)
This commit is contained in:
bbedward
2026-07-26 19:43:17 -04:00
committed by dms-ci[bot]
parent 6ba4b79039
commit 7058b00091
3 changed files with 88 additions and 10 deletions
+23 -10
View File
@@ -229,6 +229,9 @@ func (m *Manager) snapshotState() CUPSState {
func (m *Manager) Subscribe(id string) chan CUPSState { func (m *Manager) Subscribe(id string) chan CUPSState {
ch := make(chan CUPSState, 64) ch := make(chan CUPSState, 64)
m.subLifecycleMu.Lock()
defer m.subLifecycleMu.Unlock()
wasEmpty := true wasEmpty := true
m.subscribers.Range(func(key string, ch chan CUPSState) bool { m.subscribers.Range(func(key string, ch chan CUPSState) bool {
wasEmpty = false wasEmpty = false
@@ -237,19 +240,25 @@ func (m *Manager) Subscribe(id string) chan CUPSState {
m.subscribers.Store(id, ch) m.subscribers.Store(id, ch)
if wasEmpty && m.subscription != nil { if !wasEmpty || m.subscription == nil {
if err := m.subscription.Start(); err != nil { return ch
log.Warnf("[CUPS] Failed to start subscription manager: %v", err)
} else {
m.eventWG.Add(1)
go m.eventHandler()
}
} }
if err := m.subscription.Start(); err != nil {
log.Warnf("[CUPS] Failed to start subscription manager: %v", err)
return ch
}
m.eventWG.Add(1)
go m.eventHandler()
return ch return ch
} }
func (m *Manager) Unsubscribe(id string) { func (m *Manager) Unsubscribe(id string) {
m.subLifecycleMu.Lock()
defer m.subLifecycleMu.Unlock()
if val, ok := m.subscribers.LoadAndDelete(id); ok { if val, ok := m.subscribers.LoadAndDelete(id); ok {
close(val) close(val)
} }
@@ -260,18 +269,22 @@ func (m *Manager) Unsubscribe(id string) {
return false return false
}) })
if isEmpty && m.subscription != nil { if !isEmpty || m.subscription == nil {
m.subscription.Stop() return
m.eventWG.Wait()
} }
m.subscription.Stop()
m.eventWG.Wait()
} }
func (m *Manager) Close() { func (m *Manager) Close() {
close(m.stopChan) close(m.stopChan)
m.subLifecycleMu.Lock()
if m.subscription != nil { if m.subscription != nil {
m.subscription.Stop() m.subscription.Stop()
} }
m.subLifecycleMu.Unlock()
m.eventWG.Wait() m.eventWG.Wait()
m.notifierWg.Wait() m.notifierWg.Wait()
+64
View File
@@ -1,6 +1,9 @@
package cups package cups
import ( import (
"errors"
"fmt"
"sync"
"testing" "testing"
mocks_cups "github.com/AvengeMedia/DankMaterialShell/core/internal/mocks/cups" mocks_cups "github.com/AvengeMedia/DankMaterialShell/core/internal/mocks/cups"
@@ -75,6 +78,67 @@ func TestManager_Subscribe(t *testing.T) {
assert.Equal(t, 0, count) assert.Equal(t, 0, count)
} }
// mirrors the real managers: eventChan guarded by mu, conn/running deliberately
// unsynchronized so overlapping Start/Stop trips the race detector
type stubSubscription struct {
mu sync.Mutex
events chan SubscriptionEvent
conn *int
running bool
}
func (s *stubSubscription) Start() error {
if s.running {
return errors.New("already running")
}
s.running = true
s.mu.Lock()
s.events = make(chan SubscriptionEvent)
s.mu.Unlock()
v := 0
s.conn = &v
*s.conn++
return nil
}
func (s *stubSubscription) Stop() {
if !s.running {
return
}
s.running = false
s.conn = nil
s.mu.Lock()
close(s.events)
s.mu.Unlock()
}
func (s *stubSubscription) Events() <-chan SubscriptionEvent {
s.mu.Lock()
defer s.mu.Unlock()
return s.events
}
func TestManager_SubscribeUnsubscribeRace(t *testing.T) {
m := NewTestManager(mocks_cups.NewMockCUPSClientInterface(t), nil)
m.subscription = &stubSubscription{}
var wg sync.WaitGroup
for i := range 8 {
wg.Go(func() {
id := fmt.Sprintf("client-%d", i)
for range 50 {
m.Subscribe(id)
m.Unsubscribe(id)
}
})
}
wg.Wait()
}
func TestManager_Close(t *testing.T) { func TestManager_Close(t *testing.T) {
mockClient := mocks_cups.NewMockCUPSClientInterface(t) mockClient := mocks_cups.NewMockCUPSClientInterface(t)
+1
View File
@@ -79,6 +79,7 @@ type Manager struct {
client CUPSClientInterface client CUPSClientInterface
pkHelper PkHelper pkHelper PkHelper
subscription SubscriptionManagerInterface subscription SubscriptionManagerInterface
subLifecycleMu sync.Mutex
stateMutex sync.RWMutex stateMutex sync.RWMutex
subscribers syncmap.Map[string, chan CUPSState] subscribers syncmap.Map[string, chan CUPSState]
stopChan chan struct{} stopChan chan struct{}