b2fe39a fix(core): order proxy checks and destructor claims with the wire write · wlturbo

M client.go +7 -1


@@ -247,6 +247,12 @@ func (d *Display) SendRequest(objectID uint32, opcode uint16, args ...interface{
247	247
248	248		// SendRequestWithFDs sends a request with file descriptors
249	249		func (d *Display) SendRequestWithFDs(objectID uint32, opcode uint16, fds []int, args ...interface{}) error {
	250	+	return d.sendRequest(objectID, opcode, fds, nil, args)
	251	+	}
	252	+
	253	+	// sendRequest marshals a request, then runs guard and writes the message
	254	+	// under the send lock. Guard failures and marshaling errors write nothing.
	255	+	func (d *Display) sendRequest(objectID uint32, opcode uint16, fds []int, guard func() error, args []interface{}) error {
250	256		// Get buffer from pool
251	257		buf := bufferPool.Get().(*bytes.Buffer)
252	258		defer func() {

@@ -280,7 +286,7 @@ func (d *Display) SendRequestWithFDs(objectID uint32, opcode uint16, fds []int,
280	286		copy(data[0:8], header)
281	287
282	288		// Send message with optional file descriptors
283		-	return d.sendmsgWithFDs(data, fds)
	289	+	return d.sendmsgWithFDs(data, fds, guard)
284	290		}
285	291
286	292		// marshalArg marshals a single argument

 // SendRequestWithFDs sends a request with file descriptors
 func (d *Display) SendRequestWithFDs(objectID uint32, opcode uint16, fds []int, args ...interface{}) error {
+	return d.sendRequest(objectID, opcode, fds, nil, args)
+}
+// sendRequest marshals a request, then runs guard and writes the message
+// under the send lock. Guard failures and marshaling errors write nothing.
+func (d *Display) sendRequest(objectID uint32, opcode uint16, fds []int, guard func() error, args []interface{}) error {
 	// Get buffer from pool
 	buf := bufferPool.Get().(*bytes.Buffer)
 	defer func() {
 	copy(data[0:8], header)
 	// Send message with optional file descriptors
-	return d.sendmsgWithFDs(data, fds)
+	return d.sendmsgWithFDs(data, fds, guard)
 }
 // marshalArg marshals a single argument

M core.go +17 -16


@@ -58,20 +58,20 @@ func NewContext(display Display) Context {
58	58		}
59	59		}
60	60
61		-	// SendRequest sends a request through the context
	61	+	// SendRequest sends a request through the context. The proxy is validated
	62	+	// under the connection send lock, so no request reaches the wire after a
	63	+	// destructor for the same proxy.
62	64		func (c *Context) SendRequest(proxy Proxy, opcode uint32, args ...interface{}) error {
63		-	if err := c.CheckProxy(proxy); err != nil {
64		-	return err
65		-	}
66		-	return c.display.SendRequest(proxy.ID(), uint16(opcode), args...)
	65	+	return c.SendRequestWithFDs(proxy, opcode, nil, args...)
67	66		}
68	67
69		-	// SendRequestWithFDs sends a request with file descriptors through the context
	68	+	// SendRequestWithFDs sends a request with file descriptors through the context.
	69	+	// On error the FDs remain owned by the caller.
70	70		func (c *Context) SendRequestWithFDs(proxy Proxy, opcode uint32, fds []int, args ...interface{}) error {
71	71		if err := c.CheckProxy(proxy); err != nil {
72	72		return err
73	73		}
74		-	return c.display.SendRequestWithFDs(proxy.ID(), uint16(opcode), fds, args...)
	74	+	return c.display.sendRequest(proxy.ID(), uint16(opcode), fds, func() error { return c.CheckProxy(proxy) }, args)
75	75		}
76	76
77	77		// CheckProxy rejects stale and foreign proxies before any bytes are written.

@@ -90,21 +90,22 @@ func (c *Context) CheckProxy(proxy Proxy) error {
90	90		}
91	91
92	92		// SendDestructor sends a destructor request exactly once. The proxy is
93		-	// unregistered atomically before sending, so concurrent callers cannot both
94		-	// write it; the losers receive an error and nothing reaches the wire.
	93	+	// claimed (unregistered) under the connection send lock, after marshaling
	94	+	// and immediately before the write: concurrent destructors and later requests
	95	+	// on the proxy fail without writing. A marshaling error leaves the proxy
	96	+	// registered. A write error after the claim leaves the proxy unregistered;
	97	+	// such an error means the connection is broken and must be closed.
95	98		func (c *Context) SendDestructor(proxy Proxy, opcode uint32, args ...interface{}) error {
96		-	if err := c.claimDestroy(proxy); err != nil {
97		-	return err
98		-	}
99		-	return c.display.SendRequest(proxy.ID(), uint16(opcode), args...)
	99	+	return c.SendDestructorWithFDs(proxy, opcode, nil, args...)
100	100		}
101	101
102		-	// SendDestructorWithFDs is SendDestructor for requests carrying FDs.
	102	+	// SendDestructorWithFDs is SendDestructor for requests carrying FDs. On error
	103	+	// the FDs remain owned by the caller.
103	104		func (c *Context) SendDestructorWithFDs(proxy Proxy, opcode uint32, fds []int, args ...interface{}) error {
104		-	if err := c.claimDestroy(proxy); err != nil {
	105	+	if err := c.CheckProxy(proxy); err != nil {
105	106		return err
106	107		}
107		-	return c.display.SendRequestWithFDs(proxy.ID(), uint16(opcode), fds, args...)
	108	+	return c.display.sendRequest(proxy.ID(), uint16(opcode), fds, func() error { return c.claimDestroy(proxy) }, args)
108	109		}
109	110
110	111		func (c *Context) claimDestroy(proxy Proxy) error {

 	}
 }
-// SendRequest sends a request through the context
+// SendRequest sends a request through the context. The proxy is validated
+// under the connection send lock, so no request reaches the wire after a
+// destructor for the same proxy.
 func (c *Context) SendRequest(proxy Proxy, opcode uint32, args ...interface{}) error {
-	if err := c.CheckProxy(proxy); err != nil {
+	return c.SendRequestWithFDs(proxy, opcode, nil, args...)
-		return err
-	}
-	return c.display.SendRequest(proxy.ID(), uint16(opcode), args...)
 }
-// SendRequestWithFDs sends a request with file descriptors through the context
+// SendRequestWithFDs sends a request with file descriptors through the context.
+// On error the FDs remain owned by the caller.
 func (c *Context) SendRequestWithFDs(proxy Proxy, opcode uint32, fds []int, args ...interface{}) error {
 	if err := c.CheckProxy(proxy); err != nil {
 		return err
 	}
-	return c.display.SendRequestWithFDs(proxy.ID(), uint16(opcode), fds, args...)
+	return c.display.sendRequest(proxy.ID(), uint16(opcode), fds, func() error { return c.CheckProxy(proxy) }, args)
 }
 // CheckProxy rejects stale and foreign proxies before any bytes are written.
 }
 // SendDestructor sends a destructor request exactly once. The proxy is
-// unregistered atomically before sending, so concurrent callers cannot both
+// claimed (unregistered) under the connection send lock, after marshaling
-// write it; the losers receive an error and nothing reaches the wire.
+// and immediately before the write: concurrent destructors and later requests
+// on the proxy fail without writing. A marshaling error leaves the proxy
+// registered. A write error after the claim leaves the proxy unregistered;
+// such an error means the connection is broken and must be closed.
 func (c *Context) SendDestructor(proxy Proxy, opcode uint32, args ...interface{}) error {
-	if err := c.claimDestroy(proxy); err != nil {
+	return c.SendDestructorWithFDs(proxy, opcode, nil, args...)
-		return err
-	}
-	return c.display.SendRequest(proxy.ID(), uint16(opcode), args...)
 }
-// SendDestructorWithFDs is SendDestructor for requests carrying FDs.
+// SendDestructorWithFDs is SendDestructor for requests carrying FDs. On error
+// the FDs remain owned by the caller.
 func (c *Context) SendDestructorWithFDs(proxy Proxy, opcode uint32, fds []int, args ...interface{}) error {
-	if err := c.claimDestroy(proxy); err != nil {
+	if err := c.CheckProxy(proxy); err != nil {
 		return err
 	}
-	return c.display.SendRequestWithFDs(proxy.ID(), uint16(opcode), fds, args...)
+	return c.display.sendRequest(proxy.ID(), uint16(opcode), fds, func() error { return c.claimDestroy(proxy) }, args)
 }
 func (c *Context) claimDestroy(proxy Proxy) error {

M protocol/core/core_test.go +69 -16


@@ -440,34 +440,87 @@ func TestConcurrentDestroySendsOnce(t *testing.T) {
440	440		}
441	441		defer d.Close()
442	442		receiveRequest(t, p)
443		-	for round := 0; round < 200; round++ {
	443	+	const rounds, destroyers, committers = 100, 4, 4
	444	+	for round := 0; round < rounds; round++ {
444	445		surface := core.NewSurface(d.Context())
445	446		surface.SetID(d.AllocateID())
446	447		d.Context().Register(surface)
447		-	const callers = 8
448	448		start := make(chan struct{})
449		-	results := make(chan error, callers)
450		-	for i := 0; i < callers; i++ {
451		-	go func() {
452		-	<-start
453		-	results <- surface.Destroy()
454		-	}()
	449	+	destroyed := make(chan error, destroyers)
	450	+	committed := make(chan error, committers)
	451	+	for i := 0; i < destroyers; i++ {
	452	+	go func() { <-start; destroyed <- surface.Destroy() }()
	453	+	}
	454	+	for i := 0; i < committers; i++ {
	455	+	go func() { <-start; committed <- surface.Commit() }()
455	456		}
456	457		close(start)
457		-	ok := 0
458		-	for i := 0; i < callers; i++ {
459		-	if <-results == nil {
460		-	ok++
	458	+	okDestroy, okCommit := 0, 0
	459	+	for i := 0; i < destroyers; i++ {
	460	+	if <-destroyed == nil {
	461	+	okDestroy++
	462	+	}
	463	+	}
	464	+	for i := 0; i < committers; i++ {
	465	+	if <-committed == nil {
	466	+	okCommit++
461	467		}
462	468		}
463		-	if ok != 1 {
464		-	t.Fatalf("round %d: %d destroys succeeded, want 1", round, ok)
	469	+	if okDestroy != 1 {
	470	+	t.Fatalf("round %d: %d destroys succeeded, want 1", round, okDestroy)
	471	+	}
	472	+	// Decode every frame: exactly okCommit commits (opcode 6), then one
	473	+	// destroy (opcode 0) for this surface, and nothing after it.
	474	+	frames := readFrames(t, p, okCommit+1)
	475	+	for i, f := range frames {
	476	+	if f.id != surface.ID() {
	477	+	t.Fatalf("round %d: frame %d targets object %d", round, i, f.id)
	478	+	}
	479	+	last := i == len(frames)-1
	480	+	if last && f.opcode != 0 \|\| !last && f.opcode != 6 {
	481	+	t.Fatalf("round %d: frame %d opcode %d (last=%v)", round, i, f.opcode, last)
	482	+	}
465	483		}
466		-	receiveRequest(t, p)
467	484		p.SetReadDeadline(time.Now().Add(2 * time.Millisecond))
468	485		if n, err := p.Read(make([]byte, 8)); n != 0 {
469		-	t.Fatalf("round %d: extra destroy on wire (%d bytes, %v)", round, n, err)
	486	+	t.Fatalf("round %d: request after destroy (%d bytes, %v)", round, n, err)
470	487		}
471	488		p.SetReadDeadline(time.Time{})
472	489		}
473	490		}
	491	+
	492	+	type wireFrame struct {
	493	+	id uint32
	494	+	opcode uint16
	495	+	}
	496	+
	497	+	func readFrames(t testing.T, p net.UnixConn, n int) []wireFrame {
	498	+	t.Helper()
	499	+	p.SetReadDeadline(testDeadline())
	500	+	defer p.SetReadDeadline(time.Time{})
	501	+	var out []wireFrame
	502	+	var buf []byte
	503	+	tmp := make([]byte, 4096)
	504	+	for len(out) < n {
	505	+	for len(buf) >= 8 {
	506	+	size := int(binary.LittleEndian.Uint32(buf[4:8]) >> 16)
	507	+	if size < 8 \|\| len(buf) < size {
	508	+	break
	509	+	}
	510	+	out = append(out, wireFrame{binary.LittleEndian.Uint32(buf[0:4]), uint16(binary.LittleEndian.Uint32(buf[4:8]))})
	511	+	buf = buf[size:]
	512	+	}
	513	+	if len(out) >= n {
	514	+	break
	515	+	}
	516	+	m, err := p.Read(tmp)
	517	+	if err != nil {
	518	+	t.Fatalf("read frames: got %d of %d: %v", len(out), n, err)
	519	+	}
	520	+	buf = append(buf, tmp[:m]...)
	521	+	}
	522	+	if len(out) != n \|\| len(buf) != 0 {
	523	+	t.Fatalf("got %d frames (+%d bytes), want %d", len(out), len(buf), n)
	524	+	}
	525	+	return out
	526	+	}


@@ -440,34 +440,87 @@ func TestConcurrentDestroySendsOnce(t *testing.T) {
440		}	440		}
441		defer d.Close()	441		defer d.Close()
442		receiveRequest(t, p)	442		receiveRequest(t, p)
443	-	for round := 0; round < 200; round++ {	443	+	const rounds, destroyers, committers = 100, 4, 4
			444	+	for round := 0; round < rounds; round++ {
444		surface := core.NewSurface(d.Context())	445		surface := core.NewSurface(d.Context())
445		surface.SetID(d.AllocateID())	446		surface.SetID(d.AllocateID())
446		d.Context().Register(surface)	447		d.Context().Register(surface)
447	-	const callers = 8
448		start := make(chan struct{})	448		start := make(chan struct{})
449	-	results := make(chan error, callers)	449	+	destroyed := make(chan error, destroyers)
450	-	for i := 0; i < callers; i++ {	450	+	committed := make(chan error, committers)
451	-	go func() {	451	+	for i := 0; i < destroyers; i++ {
452	-	<-start	452	+	go func() { <-start; destroyed <- surface.Destroy() }()
453	-	results <- surface.Destroy()	453	+	}
454	-	}()	454	+	for i := 0; i < committers; i++ {
			455	+	go func() { <-start; committed <- surface.Commit() }()
455		}	456		}
456		close(start)	457		close(start)
457	-	ok := 0	458	+	okDestroy, okCommit := 0, 0
458	-	for i := 0; i < callers; i++ {	459	+	for i := 0; i < destroyers; i++ {
459	-	if <-results == nil {	460	+	if <-destroyed == nil {
460	-	ok++	461	+	okDestroy++
			462	+	}
			463	+	}
			464	+	for i := 0; i < committers; i++ {
			465	+	if <-committed == nil {
			466	+	okCommit++
461		}	467		}
462		}	468		}
463	-	if ok != 1 {	469	+	if okDestroy != 1 {
464	-	t.Fatalf("round %d: %d destroys succeeded, want 1", round, ok)	470	+	t.Fatalf("round %d: %d destroys succeeded, want 1", round, okDestroy)
			471	+	}
			472	+	// Decode every frame: exactly okCommit commits (opcode 6), then one
			473	+	// destroy (opcode 0) for this surface, and nothing after it.
			474	+	frames := readFrames(t, p, okCommit+1)
			475	+	for i, f := range frames {
			476	+	if f.id != surface.ID() {
			477	+	t.Fatalf("round %d: frame %d targets object %d", round, i, f.id)
			478	+	}
			479	+	last := i == len(frames)-1
			480	+	if last && f.opcode != 0 \|\| !last && f.opcode != 6 {
			481	+	t.Fatalf("round %d: frame %d opcode %d (last=%v)", round, i, f.opcode, last)
			482	+	}
465		}	483		}
466	-	receiveRequest(t, p)
467		p.SetReadDeadline(time.Now().Add(2 * time.Millisecond))	484		p.SetReadDeadline(time.Now().Add(2 * time.Millisecond))
468		if n, err := p.Read(make([]byte, 8)); n != 0 {	485		if n, err := p.Read(make([]byte, 8)); n != 0 {
469	-	t.Fatalf("round %d: extra destroy on wire (%d bytes, %v)", round, n, err)	486	+	t.Fatalf("round %d: request after destroy (%d bytes, %v)", round, n, err)
470		}	487		}
471		p.SetReadDeadline(time.Time{})	488		p.SetReadDeadline(time.Time{})
472		}	489		}
473		}	490		}
			491	+
			492	+	type wireFrame struct {
			493	+	id uint32
			494	+	opcode uint16
			495	+	}
			496	+
			497	+	func readFrames(t testing.T, p net.UnixConn, n int) []wireFrame {
			498	+	t.Helper()
			499	+	p.SetReadDeadline(testDeadline())
			500	+	defer p.SetReadDeadline(time.Time{})
			501	+	var out []wireFrame
			502	+	var buf []byte
			503	+	tmp := make([]byte, 4096)
			504	+	for len(out) < n {
			505	+	for len(buf) >= 8 {
			506	+	size := int(binary.LittleEndian.Uint32(buf[4:8]) >> 16)
			507	+	if size < 8 \|\| len(buf) < size {
			508	+	break
			509	+	}
			510	+	out = append(out, wireFrame{binary.LittleEndian.Uint32(buf[0:4]), uint16(binary.LittleEndian.Uint32(buf[4:8]))})
			511	+	buf = buf[size:]
			512	+	}
			513	+	if len(out) >= n {
			514	+	break
			515	+	}
			516	+	m, err := p.Read(tmp)
			517	+	if err != nil {
			518	+	t.Fatalf("read frames: got %d of %d: %v", len(out), n, err)
			519	+	}
			520	+	buf = append(buf, tmp[:m]...)
			521	+	}
			522	+	if len(out) != n \|\| len(buf) != 0 {
			523	+	t.Fatalf("got %d frames (+%d bytes), want %d", len(out), len(buf), n)
			524	+	}
			525	+	return out
			526	+	}

M scm_linux.go +8 -1


@@ -134,13 +134,20 @@ func (d *Display) closePendingFDs() {
134	134		func CloseSentFD(fd int) error { return unix.Close(fd) }
135	135
136	136		// sendmsgWithFDs sends a message, attaching file descriptors when present.
137		-	func (d *Display) sendmsgWithFDs(buf []byte, fds []int) error {
	137	+	// A non-nil guard runs under the send lock, so its checks and the write are
	138	+	// ordered atomically with every other request on the connection.
	139	+	func (d *Display) sendmsgWithFDs(buf []byte, fds []int, guard func() error) error {
138	140		d.sendMu.Lock()
139	141		defer d.sendMu.Unlock()
140	142
141	143		if d.closed.Load() {
142	144		return net.ErrClosed
143	145		}
	146	+	if guard != nil {
	147	+	if err := guard(); err != nil {
	148	+	return err
	149	+	}
	150	+	}
144	151		if len(fds) == 0 {
145	152		n, err := d.conn.Write(buf)
146	153		if err == nil && n != len(buf) {

 func CloseSentFD(fd int) error { return unix.Close(fd) }
 // sendmsgWithFDs sends a message, attaching file descriptors when present.
-func (d *Display) sendmsgWithFDs(buf []byte, fds []int) error {
+// A non-nil guard runs under the send lock, so its checks and the write are
+// ordered atomically with every other request on the connection.
+func (d *Display) sendmsgWithFDs(buf []byte, fds []int, guard func() error) error {
 	d.sendMu.Lock()
 	defer d.sendMu.Unlock()
 	if d.closed.Load() {
 		return net.ErrClosed
 	}
+	if guard != nil {
+		if err := guard(); err != nil {
+			return err
+		}
+	}
 	if len(fds) == 0 {
 		n, err := d.conn.Write(buf)
 		if err == nil && n != len(buf) {