Repository: godbus/dbus (go)

Issue:
Unsafe to close channel after calling Conn.RemoveSignal
If one part of my program is no longer interested in receiving D-Bus signals, it can call `RemoveSignal` to unregister the channel it provided.  At this point it would make sense to close the channel to release its resources.

However, if the channel is unbuffered or the buffer has filled up, this can lead to a panic.  While the RWLock should prevent new signals from being delivered to the channel, we could have goroutines waiting to send old signals on the channel:

https://github.com/godbus/dbus/blob/master/default_handler.go#L263-L269

It looks like the only thing that will cause these goroutines to clear is `defaultSignalHandler.Terminate`, which won't be called until the program is exiting.

The only way to avoid this seems to be to choose a channel buffer large enough that no goroutines will be spawned to handle signal broadcasts in the time it takes to acquire the write lock.  That's going to give a program that works most of the time, but may panic under load.

Pull request:
Ensure no signals will be sent on a channel after RemoveSignal is called
This fixes #158.

At present, if a signal channel's buffer is full (or it is unbuffered with no reader), a goroutine will be spun up waiting for the channel to become ready.  These goroutines will persist even after a `RemoveSignal` call, and will panic if the underlying channel is closed.

This PR pushes the wait group and closeChan down to being per-channel rather than global to the defaultSignalHandler instance.  This way we can guarantee that any associated goroutines will exit when we call RemoveSignal for a particular channel, making it safe to close.

Patch:
diff --git a/default_handler.go b/default_handler.go
index d1ce57f..6163eee 100644
--- a/default_handler.go
+++ b/default_handler.go
@@ -234,17 +234,13 @@ func (obj *exportedIntf) isFallbackInterface() bool {
 //
 // Deprecated: this is the default value, don't use it, it will be unexported.
 func NewDefaultSignalHandler() *defaultSignalHandler {
-	return &defaultSignalHandler{
-		closeChan: make(chan struct{}),
-	}
+	return &defaultSignalHandler{}
 }
 
 type defaultSignalHandler struct {
 	mu        sync.RWMutex
-	wg        sync.WaitGroup
 	closed    bool
-	signals   []chan<- *Signal
-	closeChan chan struct{}
+	signals   []*signalChannelData
 }
 
 func (sh *defaultSignalHandler) DeliverSignal(intf, name string, signal *Signal) {
@@ -253,21 +249,8 @@ func (sh *defaultSignalHandler) DeliverSignal(intf, name string, signal *Signal)
 	if sh.closed {
 		return
 	}
-	for _, ch := range sh.signals {
-		select {
-		case ch <- signal:
-		case <-sh.closeChan:
-			return
-		default:
-			sh.wg.Add(1)
-			go func(ch chan<- *Signal) {
-				select {
-				case ch <- signal:
-				case <-sh.closeChan:
-				}
-				sh.wg.Done()
-			}(ch)
-		}
+	for _, scd := range sh.signals {
+		scd.deliver(signal)
 	}
 }
 
@@ -278,10 +261,9 @@ func (sh *defaultSignalHandler) Terminate() {
 		return
 	}
 
-	close(sh.closeChan)
-	sh.wg.Wait() // wait until all spawned goroutines return
-	for _, ch := range sh.signals {
-		close(ch)
+	for _, scd := range sh.signals {
+		scd.close()
+		close(scd.ch)
 	}
 	sh.closed = true
 	sh.signals = nil
@@ -293,7 +275,10 @@ func (sh *defaultSignalHandler) AddSignal(ch chan<- *Signal) {
 	if sh.closed {
 		return
 	}
-	sh.signals = append(sh.signals, ch)
+	sh.signals = append(sh.signals, &signalChannelData{
+		ch:   ch,
+		done: make(chan struct{}),
+	})
 }
 
 func (sh *defaultSignalHandler) RemoveSignal(ch chan<- *Signal) {
@@ -303,10 +288,41 @@ func (sh *defaultSignalHandler) RemoveSignal(ch chan<- *Signal) {
 		return
 	}
 	for i := len(sh.signals) - 1; i >= 0; i-- {
-		if ch == sh.signals[i] {
+		if ch == sh.signals[i].ch {
+			sh.signals[i].close()
 			copy(sh.signals[i:], sh.signals[i+1:])
 			sh.signals[len(sh.signals)-1] = nil
 			sh.signals = sh.signals[:len(sh.signals)-1]
 		}
 	}
 }
+
+type signalChannelData struct {
+	wg   sync.WaitGroup
+	ch   chan<- *Signal
+	done chan struct{}
+}
+
+func (scd *signalChannelData) deliver(signal *Signal) {
+	select {
+	case scd.ch <- signal:
+	case <-scd.done:
+		return
+	default:
+		scd.wg.Add(1)
+		go scd.deferredDeliver(signal)
+	}
+}
+
+func (scd *signalChannelData) deferredDeliver(signal *Signal) {
+	select {
+	case scd.ch <- signal:
+	case <-scd.done:
+	}
+	scd.wg.Done()
+}
+
+func (scd *signalChannelData) close() {
+	close(scd.done)
+	scd.wg.Wait() // wait until all spawned goroutines return
+}


Test patch:
diff --git a/conn_test.go b/conn_test.go
index 4ebf2a0..d6366c7 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -103,9 +103,9 @@ func TestRemoveSignal(t *testing.T) {
 		t.Errorf("remove signal: signals length not equal: got '%d', want '3'", len(signals))
 	}
 	signals = bus.signalHandler.(*defaultSignalHandler).signals
-	for _, bch := range signals {
-		if bch != ch2 {
-			t.Errorf("remove signal: removed signal present: got '%v', want '%v'", bch, ch2)
+	for _, scd := range signals {
+		if scd.ch != ch2 {
+			t.Errorf("remove signal: removed signal present: got '%v', want '%v'", scd.ch, ch2)
 		}
 	}
 }
@@ -172,6 +172,32 @@ func TestCloseBeforeSignal(t *testing.T) {
 	}
 }
 
+func TestCloseChannelAfterRemoveSignal(t *testing.T) {
+	bus, err := NewConn(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Add an unbuffered signal channel
+	ch := make(chan *Signal)
+	bus.Signal(ch)
+
+	// Send a signal
+	msg := &Message{
+		Type: TypeSignal,
+		Headers: map[HeaderField]Variant{
+			FieldInterface: MakeVariant("foo.bar"),
+			FieldMember:    MakeVariant("bar"),
+			FieldPath:      MakeVariant(ObjectPath("/baz")),
+		},
+	}
+	bus.handleSignal(msg)
+
+	// Remove and close the signal channel
+	bus.RemoveSignal(ch)
+	close(ch)
+}
+
 func TestAddAndRemoveMatchSignal(t *testing.T) {
 	conn, err := SessionBusPrivate()
 	if err != nil {