diff --git a/internal/vzbridge/vmnet_interface.m b/internal/vzbridge/vmnet_interface.m index d1e14b9b..20bf78f2 100644 --- a/internal/vzbridge/vmnet_interface.m +++ b/internal/vzbridge/vmnet_interface.m @@ -164,7 +164,6 @@ int32_t vz_writeVmnetPacket(void *pointer, const void *buffer, uint64_t size, vo return 0; } -// vmnet limits each call to 256 KB. Use decimal KB conservatively. #if __MAC_OS_X_VERSION_MAX_ALLOWED >= 260000 static int32_t transferPackets(void *pointer, void *buffers, void *sizes, int32_t requested, bool reading, void **errorOut) { @@ -172,6 +171,7 @@ static int32_t transferPackets(void *pointer, void *buffers, void *sizes, int32_ uint64_t limit = reading ? session->readMaxPackets : session->writeMaxPackets; if (!limit) limit = 1; + // Match vmnet.packetBatchLimit so the descriptor arrays stay bounded. if (requested < 1 || requested > 32) { interfaceError(VMNET_INVALID_ARGUMENT, @"packet batch count", errorOut); return 0; @@ -179,6 +179,8 @@ static int32_t transferPackets(void *pointer, void *buffers, void *sizes, int32_ struct iovec vectors[32]; struct vmpktdesc packets[32]; uint64_t *lengths = sizes; + // Apple documents 256 KB per call. Use decimal KB, as in vmnet.packetBatchBytes. + // https://developer.apple.com/documentation/vmnet uint64_t remaining = 256000; int submitted = 0; for (; submitted < requested && submitted < limit; submitted++) { diff --git a/vmnet/batch.go b/vmnet/batch.go index be605683..9fa30bf0 100644 --- a/vmnet/batch.go +++ b/vmnet/batch.go @@ -11,7 +11,14 @@ import ( "github.com/Code-Hex/vz/v4/internal/vzbridge" ) +// Bound the native descriptor arrays to 32 packets per call. This is our batch +// size; the bridge also checks vmnet's per-interface packet limits. +// See https://developer.apple.com/documentation/vmnet/vmnet_read_max_packets_key const packetBatchLimit = 32 + +// Apple documents a 256 KB limit per read/write call. Use decimal KB to keep +// this cap conservative; the documentation does not give an exact byte count. +// See https://developer.apple.com/documentation/vmnet const packetBatchBytes = 256000 // ReadPackets reads Ethernet frames into frames and returns the number read. diff --git a/vmnetsock/adapter.go b/vmnetsock/adapter.go index 17a6a890..063ae732 100644 --- a/vmnetsock/adapter.go +++ b/vmnetsock/adapter.go @@ -72,22 +72,17 @@ func New(ctx context.Context, iface *vmnet.Interface, transport Transport) (*Ada var pumps sync.WaitGroup pumps.Go(func() { - s.finish(s.forwardToInterface(iface, transport)) + s.finish(s.forwardToInterface(run, iface, transport)) }) pumps.Go(func() { s.finish(s.forwardToPeer(run, iface, transport)) }) go func() { + <-run.Done() + s.finish(run.Err()) pumps.Wait() close(s.done) }() - go func() { - select { - case <-ctx.Done(): - s.finish(ctx.Err()) - case <-s.done: - } - }() return &Adapter{state: s}, nil } @@ -116,25 +111,33 @@ func (a *Adapter) Dup() (*os.File, error) { return os.NewFile(uintptr(fd), "vmnet-peer"), nil } -func (s *adapterState) forwardToInterface(iface *vmnet.Interface, transport Transport) error { - frame := make([]byte, iface.MaxPacketSize()) +func (s *adapterState) forwardToInterface(ctx context.Context, iface *vmnet.Interface, transport Transport) error { + var reader *streamPacketReader + var frame []byte + if transport == Stream { + reader = newStreamPacketReader(s.conn, iface.MaxPacketSize()) + } else { + frame = make([]byte, iface.MaxPacketSize()) + } for { - var n int + if err := ctx.Err(); err != nil { + return err + } var err error if transport == Stream { - n, err = readStreamFrame(s.conn, frame) + var frames [][]byte + frames, err = reader.readPackets() + if err == nil { + _, err = iface.WritePackets(frames) + } } else { + var n int n, err = readDatagramFrame(s.conn, frame) - } - if err != nil { - if errors.Is(err, io.EOF) { - return nil + if err == nil { + err = iface.WritePacket(frame[:n]) } - return err } - - err = iface.WritePacket(frame[:n]) if vmnetBufferExhausted(err) { continue } @@ -149,13 +152,17 @@ func (s *adapterState) forwardToInterface(iface *vmnet.Interface, transport Tran func (s *adapterState) forwardToPeer(ctx context.Context, iface *vmnet.Interface, transport Transport) error { size := iface.MaxPacketSize() - count := min(32, max(1, 256000/size)) + count := packetBatchSize(size) frames := make([][]byte, count) for j := range frames { frames[j] = make([]byte, size) } - headers := make([][4]byte, count) - vectors := make(net.Buffers, 2*count) + var headers [][4]byte + var vectors net.Buffers + if transport == Stream { + headers = make([][4]byte, count) + vectors = make(net.Buffers, 2*count) + } for { n, err := iface.ReadPackets(ctx, frames) if err != nil { diff --git a/vmnetsock/adapter_test.go b/vmnetsock/adapter_test.go index 6f5aa4e3..115e35cc 100644 --- a/vmnetsock/adapter_test.go +++ b/vmnetsock/adapter_test.go @@ -43,10 +43,12 @@ func TestAdapterFramesAndLifetime(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release(c) }) n, err := vmnet.NewNetwork(c) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) }) a, b := newInterface(t, n), newInterface(t, n) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -146,11 +148,16 @@ func TestAdapterCancellation(t *testing.T) { if macos.Available(26, vzbridge.SDKVersion) != nil { t.Skip("requires macOS 26 SDK") } - c, _ := vmnet.NewNetworkConfiguration(vmnet.ModeHost) + c, err := vmnet.NewNetworkConfiguration(vmnet.ModeHost) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { objc.Release(c) }) n, err := vmnet.NewNetwork(c) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) }) for _, transport := range []vmnetsock.Transport{vmnetsock.Stream, vmnetsock.Datagram} { i := newInterface(t, n) ctx, cancel := context.WithCancel(context.Background()) @@ -186,10 +193,12 @@ func TestAdapterStreamTermination(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release(c) }) n, err := vmnet.NewNetwork(c) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) }) for _, test := range []struct { name string payload []byte diff --git a/vmnetsock/backpressure_test.go b/vmnetsock/backpressure_test.go index f356ca35..6cde0763 100644 --- a/vmnetsock/backpressure_test.go +++ b/vmnetsock/backpressure_test.go @@ -11,6 +11,7 @@ import ( "time" "github.com/Code-Hex/vz/v4/internal/macos" + "github.com/Code-Hex/vz/v4/internal/objc" "github.com/Code-Hex/vz/v4/internal/vzbridge" "github.com/Code-Hex/vz/v4/vmnet" ) @@ -26,10 +27,12 @@ func TestAdapterBackpressureCleanup(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release(c) }) n, err := vmnet.NewNetwork(c) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(n)) }) receiver, err := vmnet.StartInterfaceWithNetwork(n) if err != nil { t.Fatal(err) diff --git a/vmnetsock/framing.go b/vmnetsock/framing.go index a398da45..2bf4eccf 100644 --- a/vmnetsock/framing.go +++ b/vmnetsock/framing.go @@ -1,6 +1,7 @@ package vmnetsock import ( + "bufio" "encoding/binary" "fmt" "io" @@ -9,22 +10,79 @@ import ( "golang.org/x/sys/unix" ) -func readStreamFrame(reader io.Reader, frame []byte) (int, error) { - var header [4]byte - if _, err := io.ReadFull(reader, header[:]); err != nil { - return 0, err +// Apple documents a 256 KB limit per read/write call. Use the same conservative +// decimal-KB cap as vmnet.WritePackets when grouping frames for the interface. +// See https://developer.apple.com/documentation/vmnet +const packetBatchBytes = 256000 + +func packetBatchSize(size int) int { + // Match vmnet's internal 32-packet batch size, not Apple's packet limit. + return min(32, max(1, packetBatchBytes/size)) +} + +type streamPacketReader struct { + reader *bufio.Reader + frames [][]byte + consumed int + maxSize int +} + +func newStreamPacketReader(reader io.Reader, maxSize int) *streamPacketReader { + maxSize = min(maxSize, packetBatchBytes) + count := packetBatchSize(maxSize) + return &streamPacketReader{ + reader: bufio.NewReaderSize(reader, count*(maxSize+4)), + frames: make([][]byte, 0, count), + maxSize: maxSize, } +} + +// readPackets borrows views of the reader's buffer until the next call. +// Only the first packet waits for input; the rest must already be buffered. +func (r *streamPacketReader) readPackets() ([][]byte, error) { + _, _ = r.reader.Discard(r.consumed) + r.consumed = 0 + r.frames = r.frames[:0] - size := uint64(binary.BigEndian.Uint32(header[:])) - if size == 0 || size > uint64(len(frame)) { - return 0, fmt.Errorf("invalid stream frame length %d, maximum %d", size, len(frame)) + header, err := r.reader.Peek(4) + if err != nil { + if err == io.EOF && len(header) != 0 { + err = io.ErrUnexpectedEOF + } + return nil, err + } + size, err := streamFrameSize(header, r.maxSize) + if err != nil { + return nil, err + } + if _, err := r.reader.Peek(4 + size); err != nil { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + return nil, err } - _, err := io.ReadFull(reader, frame[:int(size)]) - if err == io.EOF { - err = io.ErrUnexpectedEOF + data, _ := r.reader.Peek(r.reader.Buffered()) + for len(data) >= 4 && len(r.frames) < cap(r.frames) { + size, err := streamFrameSize(data[:4], r.maxSize) + if err != nil || len(data) < 4+size { + // Leave an invalid or incomplete suffix for the next call. + break + } + end := 4 + size + r.frames = append(r.frames, data[4:end:end]) + r.consumed += end + data = data[end:] + } + return r.frames, nil +} + +func streamFrameSize(header []byte, maximum int) (int, error) { + size := uint64(binary.BigEndian.Uint32(header)) + if size == 0 || size > uint64(maximum) { + return 0, fmt.Errorf("invalid stream frame length %d, maximum %d", size, maximum) } - return int(size), err + return int(size), nil } func writeStreamFrames(writer io.Writer, frames [][]byte, headers [][4]byte, vectors net.Buffers) error { diff --git a/vmnetsock/framing_test.go b/vmnetsock/framing_test.go index ee33bfe6..672b1232 100644 --- a/vmnetsock/framing_test.go +++ b/vmnetsock/framing_test.go @@ -30,15 +30,22 @@ func TestStreamFrames(t *testing.T) { _, _ = b.Write(part) } }() - buf := make([]byte, 3) - for _, want := range []string{"abc", "de"} { - n, err := readStreamFrame(a, buf) - if err != nil || string(buf[:n]) != want { - t.Fatalf("frame %q, %v, want %q", buf[:n], err, want) + reader := newStreamPacketReader(a, 3) + var got []string + for { + frames, err := reader.readPackets() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatal(err) + } + for _, frame := range frames { + got = append(got, string(frame)) } } - if _, err := readStreamFrame(a, buf); !errors.Is(err, io.EOF) { - t.Fatalf("EOF = %v", err) + if len(got) != 2 || got[0] != "abc" || got[1] != "de" { + t.Fatalf("frames = %q", got) } }) for _, test := range []struct { @@ -63,7 +70,7 @@ func TestStreamFrames(t *testing.T) { }, } { t.Run(test.name, func(t *testing.T) { - _, err := readStreamFrame(bytes.NewReader(test.data), make([]byte, 3)) + _, err := newStreamPacketReader(bytes.NewReader(test.data), 3).readPackets() if !errors.Is(err, test.want) { t.Fatalf("read = %v", err) } @@ -73,7 +80,7 @@ func TestStreamFrames(t *testing.T) { t.Run("invalid length", func(t *testing.T) { var header [4]byte binary.BigEndian.PutUint32(header[:], length) - if _, err := readStreamFrame(bytes.NewReader(header[:]), make([]byte, 3)); err == nil { + if _, err := newStreamPacketReader(bytes.NewReader(header[:]), 3).readPackets(); err == nil { t.Fatal("invalid length accepted") } }) @@ -232,3 +239,119 @@ func TestStreamBatchFrames(t *testing.T) { t.Fatalf("short writer = %v", err) } } + +func TestStreamPacketReaderSuffix(t *testing.T) { + for _, test := range []struct { + name string + suffix []byte + want error + }{ + { + name: "EOF", + want: io.EOF, + }, + { + name: "partial header", + suffix: []byte{0, 0}, + want: io.ErrUnexpectedEOF, + }, + { + name: "partial payload", + suffix: []byte{0, 0, 0, 3, 'x'}, + want: io.ErrUnexpectedEOF, + }, + { + name: "invalid length", + suffix: []byte{0, 0, 0, 4}, + }, + } { + t.Run(test.name, func(t *testing.T) { + data := append([]byte{0, 0, 0, 3, 'a', 'b', 'c', 0, 0, 0, 1, 'd'}, test.suffix...) + reader := newStreamPacketReader(bytes.NewReader(data), 3) + frames, err := reader.readPackets() + if err != nil || len(frames) != 2 { + t.Fatalf("complete prefix = %q, %v", frames, err) + } + if string(frames[0]) != "abc" || string(frames[1]) != "d" { + t.Fatalf("frames = %q", frames) + } + for _, frame := range frames { + if cap(frame) != len(frame) { + t.Fatalf("frame capacity = %d, length = %d", cap(frame), len(frame)) + } + } + _, err = reader.readPackets() + if err == nil || test.want != nil && !errors.Is(err, test.want) { + t.Fatalf("suffix error = %v, want %v", err, test.want) + } + }) + } +} + +func TestStreamPacketReaderDoesNotWaitForNextFrame(t *testing.T) { + a, b := net.Pipe() + defer a.Close() + defer b.Close() + if err := a.SetReadDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + continueWrite := make(chan struct{}) + done := make(chan error, 1) + go func() { + if _, err := b.Write([]byte{0, 0, 0, 1, 'a', 0, 0, 0, 2, 'b'}); err != nil { + done <- err + return + } + <-continueWrite + _, err := b.Write([]byte{'c'}) + done <- err + }() + reader := newStreamPacketReader(a, 3) + frames, err := reader.readPackets() + close(continueWrite) + if err != nil || len(frames) != 1 || string(frames[0]) != "a" { + t.Fatalf("first batch = %q, %v", frames, err) + } + frames, err = reader.readPackets() + if err != nil || len(frames) != 1 || string(frames[0]) != "bc" { + t.Fatalf("second batch = %q, %v", frames, err) + } + if err := <-done; err != nil { + t.Fatal(err) + } +} + +func TestStreamPacketReaderBatchLimits(t *testing.T) { + for _, size := range []int{1, 1500, 65550, 256000} { + t.Run(fmt.Sprint(size), func(t *testing.T) { + var data bytes.Buffer + for i := range 35 { + binary.Write(&data, binary.BigEndian, uint32(size)) + data.Write(bytes.Repeat([]byte{byte(i)}, size)) + } + reader := newStreamPacketReader(&data, size) + got := 0 + for { + frames, err := reader.readPackets() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatal(err) + } + if len(frames) == 0 || len(frames) > 32 || len(frames)*size > 256000 { + t.Fatalf("batch has %d packets of %d bytes", len(frames), size) + } + for _, frame := range frames { + if len(frame) != size || !bytes.Equal(frame, bytes.Repeat([]byte{byte(got)}, size)) { + t.Fatalf("packet %d was changed", got) + } + got++ + } + } + if got != 35 { + t.Fatalf("received %d packets, want 35", got) + } + }) + } +} diff --git a/vmnetsock/lifetime_test.go b/vmnetsock/lifetime_test.go index 679b256d..3309646d 100644 --- a/vmnetsock/lifetime_test.go +++ b/vmnetsock/lifetime_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/Code-Hex/vz/v4/internal/macos" + "github.com/Code-Hex/vz/v4/internal/objc" "github.com/Code-Hex/vz/v4/internal/vzbridge" "github.com/Code-Hex/vz/v4/vmnet" "github.com/Code-Hex/vz/v4/vmnetsock" @@ -36,10 +37,12 @@ func TestAdapterDuplicateOwnership(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release(configuration) }) network, err := vmnet.NewNetwork(configuration) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(network)) }) adapter, err := vmnetsock.New(context.Background(), newInterface(t, network), transport) if err != nil { t.Fatal(err) @@ -95,10 +98,12 @@ func TestAdapterConcurrentDupAndClose(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release(config) }) network, err := vmnet.NewNetwork(config) if err != nil { t.Fatal(err) } + t.Cleanup(func() { objc.Release((*objc.Pointer)(network)) }) for _, transport := range []vmnetsock.Transport{vmnetsock.Stream, vmnetsock.Datagram} { adapter, err := vmnetsock.New(context.Background(), newInterface(t, network), transport) if err != nil { diff --git a/vmnetsock/shutdown_test.go b/vmnetsock/shutdown_test.go new file mode 100644 index 00000000..7806ac33 --- /dev/null +++ b/vmnetsock/shutdown_test.go @@ -0,0 +1,217 @@ +package vmnetsock_test + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "io" + "net" + "testing" + "time" + + "github.com/Code-Hex/vz/v4/internal/macos" + "github.com/Code-Hex/vz/v4/internal/objc" + "github.com/Code-Hex/vz/v4/internal/vzbridge" + "github.com/Code-Hex/vz/v4/vmnet" + "github.com/Code-Hex/vz/v4/vmnetsock" +) + +func TestAdapterCloseIncompleteStream(t *testing.T) { + for _, test := range []struct { + name string + suffix []byte + race bool + }{ + { + name: "header", + suffix: []byte{0, 0}, + }, + { + name: "body", + suffix: []byte{0, 0, 0, 60, 1}, + }, + { + name: "header cancellation", + suffix: []byte{0, 0}, + race: true, + }, + { + name: "body cancellation", + suffix: []byte{0, 0, 0, 60, 1}, + race: true, + }, + } { + t.Run(test.name, func(t *testing.T) { + network := newShutdownNetwork(t) + sender, receiver := newInterface(t, network), newInterface(t, network) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + adapter, err := vmnetsock.New(ctx, sender, vmnetsock.Stream) + if err != nil { + t.Fatal(err) + } + file := duplicate(t, adapter) + peer, err := net.FileConn(file) + if err != nil { + t.Fatal(err) + } + defer peer.Close() + if err := peer.SetDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + + frame := shutdownFrame(sender, 1) + wire := binary.BigEndian.AppendUint32(nil, uint32(len(frame))) + wire = append(wire, frame...) + wire = append(wire, test.suffix...) + if _, err := peer.Write(wire); err != nil { + t.Fatal(err) + } + readShutdownFrame(t, receiver, frame) + + closed := make(chan error, 1) + go func() { + closed <- adapter.Close() + }() + if test.race { + cancel() + } + select { + case err = <-closed: + if err != nil && (!test.race || !errors.Is(err, context.Canceled)) { + t.Fatalf("Close = %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Close did not interrupt an incomplete frame") + } + if next := adapter.Close(); next != err { + t.Fatalf("repeated Close = %v, first Close = %v", next, err) + } + if _, err := file.Stat(); err != nil { + t.Fatalf("Close invalidated the caller's descriptor: %v", err) + } + frame = shutdownFrame(sender, 2) + if err := sender.WritePacket(frame); err != nil { + t.Fatalf("borrowed interface write after Close: %v", err) + } + readShutdownFrame(t, receiver, frame) + }) + } +} + +func TestAdapterStreamPrefixBeforeError(t *testing.T) { + for _, test := range []struct { + name string + suffix []byte + }{ + { + name: "invalid length", + suffix: []byte{0, 0, 0, 0}, + }, + { + name: "truncated header", + suffix: []byte{0, 0}, + }, + { + name: "truncated body", + suffix: []byte{0, 0, 0, 60, 1}, + }, + } { + t.Run(test.name, func(t *testing.T) { + network := newShutdownNetwork(t) + sender, receiver := newInterface(t, network), newInterface(t, network) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + adapter, err := vmnetsock.New(ctx, sender, vmnetsock.Stream) + if err != nil { + t.Fatal(err) + } + peer, err := net.FileConn(duplicate(t, adapter)) + if err != nil { + t.Fatal(err) + } + defer peer.Close() + if err := peer.SetDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + frames := [][]byte{shutdownFrame(sender, 1), shutdownFrame(sender, 2)} + var wire []byte + for _, frame := range frames { + wire = binary.BigEndian.AppendUint32(wire, uint32(len(frame))) + wire = append(wire, frame...) + } + wire = append(wire, test.suffix...) + if _, err := peer.Write(wire); err != nil { + t.Fatal(err) + } + if err := peer.(*net.UnixConn).CloseWrite(); err != nil { + t.Fatal(err) + } + for _, frame := range frames { + readShutdownFrame(t, receiver, frame) + } + if _, err := io.Copy(io.Discard, peer); err != nil { + t.Fatalf("adapter did not close after invalid input: %v", err) + } + closed := make(chan error, 1) + go func() { + closed <- adapter.Close() + }() + select { + case err := <-closed: + if err == nil { + t.Fatal("Close lost the stream error") + } + case <-time.After(3 * time.Second): + t.Fatal("Close did not finish after invalid input") + } + }) + } +} + +func newShutdownNetwork(t *testing.T) *vmnet.Network { + t.Helper() + if macos.Available(26, vzbridge.SDKVersion) != nil { + t.Skip("requires macOS 26 SDK") + } + config, err := vmnet.NewNetworkConfiguration(vmnet.ModeHost, vmnet.WithMTU(1500)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { objc.Release(config) }) + network, err := vmnet.NewNetwork(config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { objc.Release((*objc.Pointer)(network)) }) + return network +} + +func shutdownFrame(sender *vmnet.Interface, marker byte) []byte { + frame := bytes.Repeat([]byte{marker}, 60) + copy(frame, []byte{255, 255, 255, 255, 255, 255}) + copy(frame[6:], sender.MACAddress()) + binary.BigEndian.PutUint16(frame[12:], 0x88b5) + return frame +} + +func readShutdownFrame(t *testing.T, receiver *vmnet.Interface, want []byte) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + defer cancel() + buffer := make([]byte, receiver.MaxPacketSize()) + for { + n, err := receiver.ReadPacket(ctx, buffer) + if err != nil { + t.Fatal(err) + } + if n < 14 || binary.BigEndian.Uint16(buffer[12:]) != 0x88b5 { + continue + } + if !bytes.Equal(buffer[:n], want) { + t.Fatalf("received frame = %x, want %x", buffer[:n], want) + } + return + } +}