b4a889c perf(core): frame without per-message copies and marshal without allocations · wlturbo

M benchmark_test.go +34 -0


@@ -17,6 +17,29 @@ func BenchmarkWireDispatch(b *testing.B) {
17	17		}
18	18		}
19	19
	20	+	// BenchmarkDispatchCoalesced measures framing when one read carries many
	21	+	// messages; per-frame cost must not grow with the amount buffered.
	22	+	func BenchmarkDispatchCoalesced(b *testing.B) {
	23	+	var chunk []byte
	24	+	for len(chunk)+12 <= readChunkSize {
	25	+	chunk = append(chunk, message(7, 0, []byte{1, 0, 0, 0})...)
	26	+	}
	27	+	frames := len(chunk) / 12
	28	+	b.ReportAllocs()
	29	+	for i := 0; i < b.N; i++ {
	30	+	b.StopTimer()
	31	+	d := newDisplay(&chunkConn{chunks: [][]byte{append([]byte(nil), chunk...)}})
	32	+	d.context.Register(&benchProxy{BaseProxy: BaseProxy{id: 7, context: d.context}})
	33	+	b.StartTimer()
	34	+	for j := 0; j < frames; j++ {
	35	+	if err := d.Dispatch(); err != nil {
	36	+	b.Fatal(err)
	37	+	}
	38	+	}
	39	+	}
	40	+	b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*frames), "ns/frame")
	41	+	}
	42	+
20	43		type benchProxy struct {
21	44		BaseProxy
22	45		value uint32

@@ -24,3 +47,14 @@ type benchProxy struct {
24	47
25	48		func (p *benchProxy) EventSignature(op uint16) (string, bool) { return "uint,", op == 0 }
26	49		func (p benchProxy) Dispatch(e Event) { p.value = e.Uint32() }
	50	+
	51	+	// BenchmarkSendRequest measures marshaling a typical fixed-size request.
	52	+	func BenchmarkSendRequest(b *testing.B) {
	53	+	d := newDisplay(&chunkConn{})
	54	+	b.ReportAllocs()
	55	+	for i := 0; i < b.N; i++ {
	56	+	if err := d.SendRequest(9, 0, uint32(1), int32(2), Fixed(3)); err != nil {
	57	+	b.Fatal(err)
	58	+	}
	59	+	}
	60	+	}


@@ -17,6 +17,29 @@ func BenchmarkWireDispatch(b *testing.B) {
17	}	17		}
18	}	18		}
19		19
		20	+	// BenchmarkDispatchCoalesced measures framing when one read carries many
		21	+	// messages; per-frame cost must not grow with the amount buffered.
		22	+	func BenchmarkDispatchCoalesced(b *testing.B) {
		23	+	var chunk []byte
		24	+	for len(chunk)+12 <= readChunkSize {
		25	+	chunk = append(chunk, message(7, 0, []byte{1, 0, 0, 0})...)
		26	+	}
		27	+	frames := len(chunk) / 12
		28	+	b.ReportAllocs()
		29	+	for i := 0; i < b.N; i++ {
		30	+	b.StopTimer()
		31	+	d := newDisplay(&chunkConn{chunks: [][]byte{append([]byte(nil), chunk...)}})
		32	+	d.context.Register(&benchProxy{BaseProxy: BaseProxy{id: 7, context: d.context}})
		33	+	b.StartTimer()
		34	+	for j := 0; j < frames; j++ {
		35	+	if err := d.Dispatch(); err != nil {
		36	+	b.Fatal(err)
		37	+	}
		38	+	}
		39	+	}
		40	+	b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*frames), "ns/frame")
		41	+	}
		42	+
20	type benchProxy struct {	43		type benchProxy struct {
21	BaseProxy	44		BaseProxy
22	value uint32	45		value uint32

@@ -24,3 +47,14 @@ type benchProxy struct {
24		47
25	func (p *benchProxy) EventSignature(op uint16) (string, bool) { return "uint,", op == 0 }	48		func (p *benchProxy) EventSignature(op uint16) (string, bool) { return "uint,", op == 0 }
26	func (p benchProxy) Dispatch(e Event) { p.value = e.Uint32() }	49		func (p benchProxy) Dispatch(e Event) { p.value = e.Uint32() }
		50	+
		51	+	// BenchmarkSendRequest measures marshaling a typical fixed-size request.
		52	+	func BenchmarkSendRequest(b *testing.B) {
		53	+	d := newDisplay(&chunkConn{})
		54	+	b.ReportAllocs()
		55	+	for i := 0; i < b.N; i++ {
		56	+	if err := d.SendRequest(9, 0, uint32(1), int32(2), Fixed(3)); err != nil {
		57	+	b.Fatal(err)
		58	+	}
		59	+	}
		60	+	}

M client.go +42 -32


@@ -61,11 +61,12 @@ type Display struct {
61	61		registry *Registry
62	62		context *Context // Store context once
63	63
64		-	// Connection-local receive state. rbuf holds bytes read from the socket
65		-	// that do not yet form a complete frame; recvBuf is the scratch buffer
66		-	// used for a single read. pendingFDs holds ancillary descriptors that
67		-	// arrived with those bytes and have not been consumed yet.
	64	+	// Connection-local receive state. rbuf[rpos:] holds bytes read from the
	65	+	// socket that have not been framed yet; recvBuf is the scratch buffer used
	66	+	// for a single read. pendingFDs holds ancillary descriptors that arrived
	67	+	// with those bytes and have not been consumed yet.
68	68		rbuf []byte
	69	+	rpos int
69	70		recvBuf []byte
70	71		readErr error
71	72		maxMsgLen uint32

@@ -244,8 +245,8 @@ func (d *Display) sendRequest(objectID uint32, opcode uint16, fds []int, guard f
244	245		}()
245	246
246	247		// Write header placeholder
247		-	header := make([]byte, 8)
248		-	_, _ = buf.Write(header)
	248	+	var header [HeaderSize]byte
	249	+	_, _ = buf.Write(header[:])
249	250
250	251		// Marshal arguments
251	252		for _, arg := range args {

@@ -266,30 +267,35 @@ func (d *Display) sendRequest(objectID uint32, opcode uint16, fds []int, guard f
266	267
267	268		// Update buffer with correct header
268	269		data := buf.Bytes()
269		-	copy(data[0:8], header)
	270	+	copy(data[0:8], header[:])
270	271
271	272		// Send message with optional file descriptors
272	273		return d.sendmsgWithFDs(data, fds, guard)
273	274		}
274	275
	276	+	// putUint32 appends one little-endian wire word without allocating.
	277	+	func putUint32(buf *bytes.Buffer, v uint32) {
	278	+	var w [4]byte
	279	+	binary.LittleEndian.PutUint32(w[:], v)
	280	+	_, _ = buf.Write(w[:])
	281	+	}
	282	+
275	283		// marshalArg marshals a single argument
276	284		func (d Display) marshalArg(buf bytes.Buffer, arg interface{}) error {
277	285		switch v := arg.(type) {
278	286		case uint32:
279		-	return binary.Write(buf, binary.LittleEndian, v)
	287	+	putUint32(buf, v)
280	288		case int32:
281		-	return binary.Write(buf, binary.LittleEndian, v)
	289	+	putUint32(buf, uint32(v))
282	290		case Fixed:
283		-	return binary.Write(buf, binary.LittleEndian, int32(v))
	291	+	putUint32(buf, uint32(v))
284	292		case string:
285	293		// String format: length (including null) + string + null + padding
286	294		strlen := len(v) + 1
287	295		if strlen > 0xffff-HeaderSize {
288	296		return fmt.Errorf("string too long: %d bytes", strlen)
289	297		}
290		-	if err := binary.Write(buf, binary.LittleEndian, uint32(strlen)); err != nil { // Safe: checked above
291		-	return err
292		-	}
	298	+	putUint32(buf, uint32(strlen)) // Safe: checked above
293	299		_, _ = buf.WriteString(v)
294	300		_ = buf.WriteByte(0)
295	301		// Pad to 32-bit boundary

@@ -303,9 +309,7 @@ func (d Display) marshalArg(buf bytes.Buffer, arg interface{}) error {
303	309		if arrlen > 0xffff-HeaderSize {
304	310		return fmt.Errorf("array too long: %d bytes", arrlen)
305	311		}
306		-	if err := binary.Write(buf, binary.LittleEndian, uint32(arrlen)); err != nil {
307		-	return err
308		-	}
	312	+	putUint32(buf, uint32(arrlen))
309	313		_, _ = buf.Write(v)
310	314		// Pad to 32-bit boundary
311	315		padding := (4 - (arrlen % 4)) % 4

@@ -314,12 +318,13 @@ func (d Display) marshalArg(buf bytes.Buffer, arg interface{}) error {
314	318		}
315	319		case Object:
316	320		if v != nil {
317		-	return binary.Write(buf, binary.LittleEndian, v.ID())
	321	+	putUint32(buf, v.ID())
	322	+	} else {
	323	+	putUint32(buf, 0)
318	324		}
319		-	return binary.Write(buf, binary.LittleEndian, uint32(0))
320	325		case nil:
321	326		// Null object
322		-	return binary.Write(buf, binary.LittleEndian, uint32(0))
	327	+	putUint32(buf, 0)
323	328		case uintptr:
324	329		// File descriptor as uintptr - NO placeholder in message for neurlang compatibility
325	330		// FD is ONLY sent via SCM_RIGHTS, not in the message body

@@ -393,18 +398,20 @@ func (d *Display) maxMessageSize() uint32 {
393	398		// nextFrame returns the next complete frame, reading more bytes as needed.
394	399		func (d *Display) nextFrame() (receivedFrame, error) {
395	400		for {
396		-	if len(d.rbuf) >= HeaderSize {
397		-	object, opcode, size, err := parseHeader(d.rbuf[:HeaderSize], d.maxMessageSize())
	401	+	buf := d.rbuf[d.rpos:]
	402	+	if len(buf) >= HeaderSize {
	403	+	object, opcode, size, err := parseHeader(buf[:HeaderSize], d.maxMessageSize())
398	404		if err != nil {
399	405		return receivedFrame{}, err
400	406		}
401		-	if uint32(len(d.rbuf)) >= size {
	407	+	if uint32(len(buf)) >= size {
	408	+	// The body aliases the receive buffer. It stays valid until
	409	+	// the next fill, which only happens in a later Dispatch.
402	410		frame := receivedFrame{object: object, opcode: opcode}
403	411		if size > HeaderSize {
404		-	frame.body = make([]byte, size-HeaderSize)
405		-	copy(frame.body, d.rbuf[HeaderSize:size])
	412	+	frame.body = buf[HeaderSize:size:size]
406	413		}
407		-	d.consume(size)
	414	+	d.rpos += int(size)
408	415		return frame, nil
409	416		}
410	417		}

@@ -417,11 +424,8 @@ func (d *Display) nextFrame() (receivedFrame, error) {
417	424		}
418	425		}
419	426
420		-	// consume removes the first n bytes from the receive buffer.
421		-	func (d *Display) consume(n uint32) {
422		-	rest := copy(d.rbuf, d.rbuf[n:])
423		-	d.rbuf = d.rbuf[:rest]
424		-	}
	427	+	// unread reports whether received bytes remain that have not been framed.
	428	+	func (d *Display) unread() bool { return d.rpos < len(d.rbuf) }
425	429
426	430		// fill reads one chunk from the connection into the receive buffer. Bytes that
427	431		// arrive together with an error are kept so they are still parsed before the

@@ -433,6 +437,12 @@ func (d *Display) fill() error {
433	437		if d.recvBuf == nil {
434	438		d.recvBuf = make([]byte, readChunkSize)
435	439		}
	440	+	// Compact once per read rather than once per frame, so framing cost does
	441	+	// not grow with the number of buffered messages.
	442	+	if d.rpos > 0 {
	443	+	d.rbuf = d.rbuf[:copy(d.rbuf, d.rbuf[d.rpos:])]
	444	+	d.rpos = 0
	445	+	}
436	446		n, err := d.readChunk()
437	447		if n > 0 {
438	448		d.rbuf = append(d.rbuf, d.recvBuf[:n]...)

@@ -499,7 +509,7 @@ func (d Display) prepareFrame(f receivedFrame) (Object, Event, error) {
499	509		return nil, nil, net.ErrClosed
500	510		}
501	511		if f.object == displayObjectID {
502		-	if len(d.pendingFDs) != 0 && len(d.rbuf) == 0 {
	512	+	if len(d.pendingFDs) != 0 && !d.unread() {
503	513		return nil, nil, &ProtocolError{Kind: "extra_fd", Err: ErrMalformedFrame}
504	514		}
505	515		return nil, nil, nil

@@ -545,7 +555,7 @@ func (d Display) prepareFrame(f receivedFrame) (Object, Event, error) {
545	555		}
546	556		// FDs from a single recvmsg can accompany multiple coalesced frames. Once
547	557		// all buffered bytes have been framed, surplus descriptors are not legal.
548		-	if len(d.rbuf) == 0 && len(d.pendingFDs) != 0 {
	558	+	if !d.unread() && len(d.pendingFDs) != 0 {
549	559		for _, fd := range ev.fds {
550	560		_ = fd.Close()
551	561		}

 	registry *Registry
 	context  *Context // Store context once
-	// Connection-local receive state. rbuf holds bytes read from the socket
+	// Connection-local receive state. rbuf[rpos:] holds bytes read from the
-	// that do not yet form a complete frame; recvBuf is the scratch buffer
+	// socket that have not been framed yet; recvBuf is the scratch buffer used
-	// used for a single read. pendingFDs holds ancillary descriptors that
+	// for a single read. pendingFDs holds ancillary descriptors that arrived
-	// arrived with those bytes and have not been consumed yet.
+	// with those bytes and have not been consumed yet.
 	rbuf       []byte
+	rpos       int
 	recvBuf    []byte
 	readErr    error
 	maxMsgLen  uint32
 	}()
 	// Write header placeholder
-	header := make([]byte, 8)
+	var header [HeaderSize]byte
-	_, _ = buf.Write(header)
+	_, _ = buf.Write(header[:])
 	// Marshal arguments
 	for _, arg := range args {
 	// Update buffer with correct header
 	data := buf.Bytes()
-	copy(data[0:8], header)
+	copy(data[0:8], header[:])
 	// Send message with optional file descriptors
 	return d.sendmsgWithFDs(data, fds, guard)
 }
+// putUint32 appends one little-endian wire word without allocating.
+func putUint32(buf *bytes.Buffer, v uint32) {
+	var w [4]byte
+	binary.LittleEndian.PutUint32(w[:], v)
+	_, _ = buf.Write(w[:])
+}
 // marshalArg marshals a single argument
 func (d *Display) marshalArg(buf *bytes.Buffer, arg interface{}) error {
 	switch v := arg.(type) {
 	case uint32:
-		return binary.Write(buf, binary.LittleEndian, v)
+		putUint32(buf, v)
 	case int32:
-		return binary.Write(buf, binary.LittleEndian, v)
+		putUint32(buf, uint32(v))
 	case Fixed:
-		return binary.Write(buf, binary.LittleEndian, int32(v))
+		putUint32(buf, uint32(v))
 	case string:
 		// String format: length (including null) + string + null + padding
 		strlen := len(v) + 1
 		if strlen > 0xffff-HeaderSize {
 			return fmt.Errorf("string too long: %d bytes", strlen)
 		}
-		if err := binary.Write(buf, binary.LittleEndian, uint32(strlen)); err != nil { // Safe: checked above
+		putUint32(buf, uint32(strlen)) // Safe: checked above
-			return err
-		}
 		_, _ = buf.WriteString(v)
 		_ = buf.WriteByte(0)
 		// Pad to 32-bit boundary
 		if arrlen > 0xffff-HeaderSize {
 			return fmt.Errorf("array too long: %d bytes", arrlen)
 		}
-		if err := binary.Write(buf, binary.LittleEndian, uint32(arrlen)); err != nil {
+		putUint32(buf, uint32(arrlen))
-			return err
-		}
 		_, _ = buf.Write(v)
 		// Pad to 32-bit boundary
 		padding := (4 - (arrlen % 4)) % 4
 		}
 	case Object:
 		if v != nil {
-			return binary.Write(buf, binary.LittleEndian, v.ID())
+			putUint32(buf, v.ID())
+		} else {
+			putUint32(buf, 0)
 		}
-		return binary.Write(buf, binary.LittleEndian, uint32(0))
 	case nil:
 		// Null object
-		return binary.Write(buf, binary.LittleEndian, uint32(0))
+		putUint32(buf, 0)
 	case uintptr:
 		// File descriptor as uintptr - NO placeholder in message for neurlang compatibility
 		// FD is ONLY sent via SCM_RIGHTS, not in the message body
 // nextFrame returns the next complete frame, reading more bytes as needed.
 func (d *Display) nextFrame() (receivedFrame, error) {
 	for {
-		if len(d.rbuf) >= HeaderSize {
+		buf := d.rbuf[d.rpos:]
-			object, opcode, size, err := parseHeader(d.rbuf[:HeaderSize], d.maxMessageSize())
+		if len(buf) >= HeaderSize {
+			object, opcode, size, err := parseHeader(buf[:HeaderSize], d.maxMessageSize())
 			if err != nil {
 				return receivedFrame{}, err
 			}
-			if uint32(len(d.rbuf)) >= size {
+			if uint32(len(buf)) >= size {
+				// The body aliases the receive buffer. It stays valid until
+				// the next fill, which only happens in a later Dispatch.
 				frame := receivedFrame{object: object, opcode: opcode}
 				if size > HeaderSize {
-					frame.body = make([]byte, size-HeaderSize)
+					frame.body = buf[HeaderSize:size:size]
-					copy(frame.body, d.rbuf[HeaderSize:size])
 				}
-				d.consume(size)
+				d.rpos += int(size)
 				return frame, nil
 			}
 		}
 	}
 }
-// consume removes the first n bytes from the receive buffer.
+// unread reports whether received bytes remain that have not been framed.
-func (d *Display) consume(n uint32) {
+func (d *Display) unread() bool { return d.rpos < len(d.rbuf) }
-	rest := copy(d.rbuf, d.rbuf[n:])
-	d.rbuf = d.rbuf[:rest]
-}
 // fill reads one chunk from the connection into the receive buffer. Bytes that
 // arrive together with an error are kept so they are still parsed before the
 	if d.recvBuf == nil {
 		d.recvBuf = make([]byte, readChunkSize)
 	}
+	// Compact once per read rather than once per frame, so framing cost does
+	// not grow with the number of buffered messages.
+	if d.rpos > 0 {
+		d.rbuf = d.rbuf[:copy(d.rbuf, d.rbuf[d.rpos:])]
+		d.rpos = 0
+	}
 	n, err := d.readChunk()
 	if n > 0 {
 		d.rbuf = append(d.rbuf, d.recvBuf[:n]...)
 		return nil, nil, net.ErrClosed
 	}
 	if f.object == displayObjectID {
-		if len(d.pendingFDs) != 0 && len(d.rbuf) == 0 {
+		if len(d.pendingFDs) != 0 && !d.unread() {
 			return nil, nil, &ProtocolError{Kind: "extra_fd", Err: ErrMalformedFrame}
 		}
 		return nil, nil, nil
 	}
 	// FDs from a single recvmsg can accompany multiple coalesced frames. Once
 	// all buffered bytes have been framed, surplus descriptors are not legal.
-	if len(d.rbuf) == 0 && len(d.pendingFDs) != 0 {
+	if !d.unread() && len(d.pendingFDs) != 0 {
 		for _, fd := range ev.fds {
 			_ = fd.Close()
 		}

M core.go +2 -1


@@ -45,7 +45,8 @@ type Event struct {
45	45		fdIndex int
46	46		}
47	47
48		-	// Data returns the raw event data
	48	+	// Data returns the raw event body. It aliases the connection's receive buffer
	49	+	// and is only valid until the handler returns; copy it to keep it.
49	50		func (e *Event) Data() []byte {
50	51		return e.data
51	52		}


@@ -45,7 +45,8 @@ type Event struct {
45		fdIndex int	45		fdIndex int
46		}	46		}
47			47
48	-	// Data returns the raw event data	48	+	// Data returns the raw event body. It aliases the connection's receive buffer
			49	+	// and is only valid until the handler returns; copy it to keep it.
49		func (e *Event) Data() []byte {	50		func (e *Event) Data() []byte {
50		return e.data	51		return e.data
51		}	52		}