Repository: godbus/dbus
Pull request #215 (2020-10-16 06:29:59)

sequential_handler: simplify to use less goroutines
This is an attempt to reduce the number of goroutines used by the SequentialSignalHandler implementation, which seemed to occasionally trip up the race detector in the tests.  The main changes are:

1. spin up one goroutine to handle each signal channel rather than one per signal per channel.
2. move the handling of the queue of undelivered signals into this goroutine
3. remove the locks and wait group that were serialising access to the queue, since it is now private to a single goroutine.

The third `done` channel is needed to ensure the `bufferSignals` goroutine has completed.  Without it, it's possible that the goroutine will still be stuck in the `select` call when the user closes the signal channel, leading to a panic.

I've kept the use of the list type, but it may be more effective to use a slice if we expect the number of buffered signals to be small.


Patch:
diff --git a/conn.go b/conn.go
index f959d21..664bc62 100644
--- a/conn.go
+++ b/conn.go
@@ -464,6 +464,9 @@ func (conn *Conn) Object(dest string, path ObjectPath) BusObject {
 }
 
 func (conn *Conn) sendMessageAndIfClosed(msg *Message, ifClosed func()) {
+	if msg.serial == 0 {
+		msg.serial = conn.getSerial()
+	}
 	if conn.outInt != nil {
 		conn.outInt(msg)
 	}
@@ -495,16 +498,16 @@ func (conn *Conn) send(ctx context.Context, msg *Message, ch chan *Call) *Call {
 	if ctx == nil {
 		panic("nil context")
 	}
+	if ch == nil {
+		ch = make(chan *Call, 1)
+	} else if cap(ch) == 0 {
+		panic("dbus: unbuffered channel passed to (*Conn).Send")
+	}
 
 	var call *Call
 	ctx, canceler := context.WithCancel(ctx)
 	msg.serial = conn.getSerial()
 	if msg.Type == TypeMethodCall && msg.Flags&FlagNoReplyExpected == 0 {
-		if ch == nil {
-			ch = make(chan *Call, 5)
-		} else if cap(ch) == 0 {
-			panic("dbus: unbuffered channel passed to (*Conn).Send")
-		}
 		call = new(Call)
 		call.Destination, _ = msg.Headers[FieldDestination].value.(string)
 		call.Path, _ = msg.Headers[FieldPath].value.(ObjectPath)
@@ -526,7 +529,8 @@ func (conn *Conn) send(ctx context.Context, msg *Message, ch chan *Call) *Call {
 		})
 	} else {
 		canceler()
-		call = &Call{Err: nil}
+		call = &Call{Err: nil, Done: ch}
+		ch <- call
 		conn.sendMessageAndIfClosed(msg, func() {
 			call = &Call{Err: ErrClosed}
 		})
@@ -551,7 +555,6 @@ func (conn *Conn) sendError(err error, dest string, serial uint32) {
 	}
 	msg := new(Message)
 	msg.Type = TypeError
-	msg.serial = conn.getSerial()
 	msg.Headers = make(map[HeaderField]Variant)
 	if dest != "" {
 		msg.Headers[FieldDestination] = MakeVariant(dest)
@@ -570,7 +573,6 @@ func (conn *Conn) sendError(err error, dest string, serial uint32) {
 func (conn *Conn) sendReply(dest string, serial uint32, values ...interface{}) {
 	msg := new(Message)
 	msg.Type = TypeMethodReply
-	msg.serial = conn.getSerial()
 	msg.Headers = make(map[HeaderField]Variant)
 	if dest != "" {
 		msg.Headers[FieldDestination] = MakeVariant(dest)
diff --git a/export.go b/export.go
index 1093b01..bfc5db6 100644
--- a/export.go
+++ b/export.go
@@ -175,7 +175,6 @@ func (conn *Conn) handleCall(msg *Message) {
 	if msg.Flags&FlagNoReplyExpected == 0 {
 		reply := new(Message)
 		reply.Type = TypeMethodReply
-		reply.serial = conn.getSerial()
 		reply.Headers = make(map[HeaderField]Variant)
 		if hasSender {
 			reply.Headers[FieldDestination] = msg.Headers[FieldSender]
@@ -211,7 +210,6 @@ func (conn *Conn) Emit(path ObjectPath, name string, values ...interface{}) erro
 	}
 	msg := new(Message)
 	msg.Type = TypeSignal
-	msg.serial = conn.getSerial()
 	msg.Headers = make(map[HeaderField]Variant)
 	msg.Headers[FieldInterface] = MakeVariant(iface)
 	msg.Headers[FieldMember] = MakeVariant(member)
diff --git a/object.go b/object.go
index 6dc10dc..664abb7 100644
--- a/object.go
+++ b/object.go
@@ -110,7 +110,6 @@ func (o *Object) createCall(ctx context.Context, method string, flags Flags, ch
 	method = method[i+1:]
 	msg := new(Message)
 	msg.Type = TypeMethodCall
-	msg.serial = o.conn.getSerial()
 	msg.Flags = flags & (FlagNoAutoStart | FlagNoReplyExpected)
 	msg.Headers = make(map[HeaderField]Variant)
 	msg.Headers[FieldPath] = MakeVariant(o.path)
@@ -123,47 +122,7 @@ func (o *Object) createCall(ctx context.Context, method string, flags Flags, ch
 	if len(args) > 0 {
 		msg.Headers[FieldSignature] = MakeVariant(SignatureOf(args...))
 	}
-	if msg.Flags&FlagNoReplyExpected == 0 {
-		if ch == nil {
-			ch = make(chan *Call, 1)
-		} else if cap(ch) == 0 {
-			panic("dbus: unbuffered channel passed to (*Object).Go")
-		}
-		ctx, cancel := context.WithCancel(ctx)
-		call := &Call{
-			Destination: o.dest,
-			Path:        o.path,
-			Method:      method,
-			Args:        args,
-			Done:        ch,
-			ctxCanceler: cancel,
-			ctx:         ctx,
-		}
-		o.conn.calls.track(msg.serial, call)
-		o.conn.sendMessageAndIfClosed(msg, func() {
-			o.conn.calls.handleSendError(msg, ErrClosed)
-			cancel()
-		})
-		go func() {
-			<-ctx.Done()
-			o.conn.calls.handleSendError(msg, ctx.Err())
-		}()
-
-		return call
-	}
-	done := make(chan *Call, 1)
-	call := &Call{
-		Err:  nil,
-		Done: done,
-	}
-	defer func() {
-		call.Done <- call
-		close(done)
-	}()
-	o.conn.sendMessageAndIfClosed(msg, func() {
-		call.Err = ErrClosed
-	})
-	return call
+	return o.conn.SendWithContext(ctx, msg, ch)
 }
 
 // GetProperty calls org.freedesktop.DBus.Properties.Get on the given
diff --git a/sequential_handler.go b/sequential_handler.go
index 3a8cd71..ef2fcdb 100644
--- a/sequential_handler.go
+++ b/sequential_handler.go
@@ -1,7 +1,6 @@
 package dbus
 
 import (
-	"container/list"
 	"sync"
 )
 
@@ -51,11 +50,7 @@ func (sh *sequentialSignalHandler) AddSignal(ch chan<- *Signal) {
 	if sh.closed {
 		return
 	}
-	sh.signals = append(sh.signals, &sequentialSignalChannelData{
-		queue: list.New(),
-		ch:    ch,
-		done:  make(chan struct{}),
-	})
+	sh.signals = append(sh.signals, newSequentialSignalChannelData(ch))
 }
 
 func (sh *sequentialSignalHandler) RemoveSignal(ch chan<- *Signal) {
@@ -75,46 +70,56 @@ func (sh *sequentialSignalHandler) RemoveSignal(ch chan<- *Signal) {
 }
 
 type sequentialSignalChannelData struct {
-	stateLock sync.Mutex
-	writeLock sync.Mutex
-	queue     *list.List
-	wg        sync.WaitGroup
-	ch        chan<- *Signal
-	done      chan struct{}
+	ch   chan<- *Signal
+	in   chan *Signal
+	done chan struct{}
 }
 
-func (scd *sequentialSignalChannelData) deliver(signal *Signal) {
-	// Avoid blocking the main DBus message processing routine;
-	// queue signal to be dispatched later.
-	scd.stateLock.Lock()
-	scd.queue.PushBack(signal)
-	scd.stateLock.Unlock()
-
-	scd.wg.Add(1)
-	go scd.deferredDeliver()
+func newSequentialSignalChannelData(ch chan<- *Signal) *sequentialSignalChannelData {
+	scd := &sequentialSignalChannelData{
+		ch:   ch,
+		in:   make(chan *Signal),
+		done: make(chan struct{}),
+	}
+	go scd.bufferSignals()
+	return scd
 }
 
-func (scd *sequentialSignalChannelData) deferredDeliver() {
-	defer scd.wg.Done()
-
-	// Ensure only one goroutine is in this section at once, to
-	// make sure signals are sent over ch in the order they
-	// are in the queue.
-	scd.writeLock.Lock()
-	defer scd.writeLock.Unlock()
-
-	scd.stateLock.Lock()
-	elem := scd.queue.Front()
-	scd.queue.Remove(elem)
-	scd.stateLock.Unlock()
-
-	select {
-	case scd.ch <- elem.Value.(*Signal):
-	case <-scd.done:
+func (scd *sequentialSignalChannelData) bufferSignals() {
+	defer close(scd.done)
+
+	// Ensure that signals are delivered to scd.ch in the same
+	// order they are received from scd.in.
+	var queue []*Signal
+	for {
+		if len(queue) == 0 {
+			signal, ok := <- scd.in
+			if !ok {
+				return
+			}
+			queue = append(queue, signal)
+		}
+		select {
+		case scd.ch <- queue[0]:
+			copy(queue, queue[1:])
+			queue[len(queue)-1] = nil
+			queue = queue[:len(queue)-1]
+		case signal, ok := <-scd.in:
+			if !ok {
+				return
+			}
+			queue = append(queue, signal)
+		}
 	}
 }
 
+func (scd *sequentialSignalChannelData) deliver(signal *Signal) {
+	scd.in <- signal
+}
+
 func (scd *sequentialSignalChannelData) close() {
-	close(scd.done)
-	scd.wg.Wait() // wait until all spawned goroutines return
+	close(scd.in)
+	// Ensure that bufferSignals() has exited and won't attempt
+	// any future sends on scd.ch
+	<-scd.done
 }


Test patch:
diff --git a/conn_test.go b/conn_test.go
index ff23b71..8ac32b8 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -461,7 +461,6 @@ process:
 				t.Logf("Processing reply to .State(), returning state = %v", state)
 				reply := new(Message)
 				reply.Type = TypeMethodReply
-				reply.serial = srv.getSerial()
 				reply.Headers = make(map[HeaderField]Variant)
 				reply.Headers[FieldDestination] = msg.Headers[FieldSender]
 				reply.Headers[FieldReplySerial] = MakeVariant(msg.serial)