From 9bc603c3c5b86516c6dd1d5aec903fe60ea39299 Mon Sep 17 00:00:00 2001 From: Fabian Fromwald Date: Fri, 5 Dec 2025 14:37:02 +0100 Subject: [PATCH 1/4] Use quic-go fork with priority scheduling --- go.mod | 5 ++++- go.sum | 8 ++++---- quic_connection.go | 8 ++++++++ send_flow.go | 5 ++++- session.go | 3 +++ 5 files changed, 23 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index 0326a11..07192cc 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/mengelbart/roq -go 1.24 +go 1.24.0 require ( github.com/mengelbart/qlog v0.0.0-20250304144032-13728b0b6fae @@ -20,5 +20,8 @@ require ( golang.org/x/crypto v0.41.0 // indirect golang.org/x/net v0.43.0 // indirect golang.org/x/sys v0.35.0 // indirect + golang.org/x/time v0.14.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) + +replace github.com/quic-go/quic-go v0.57.1 => github.com/Willi-42/quic-go v0.7.1 diff --git a/go.sum b/go.sum index c82a269..40dbcb9 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/Willi-42/quic-go v0.7.1 h1:6Rjzjo1xmFbiGJ8Xib+CpB9imeJTEhmHs8jgGm1uceU= +github.com/Willi-42/quic-go v0.7.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -70,8 +72,6 @@ github.com/pion/webrtc/v3 v3.2.28 h1:ienStxZ6HcjtH2UlmnFpMM0loENiYjaX437uIUpQSKo github.com/pion/webrtc/v3 v3.2.28/go.mod h1:PNRCEuQlibrmuBhOTnol9j6KkIbUG11aHLEfNpUYey0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/quic-go/quic-go v0.57.1 h1:25KAAR9QR8KZrCZRThWMKVAwGoiHIrNbT72ULHTuI10= -github.com/quic-go/quic-go v0.57.1/go.mod h1:ly4QBAjHA2VhdnxhojRsCUOeJwKYg+taDlos92xb1+s= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/sclevine/agouti v3.0.0+incompatible/go.mod h1:b4WX9W9L1sfQKXeJf1mUTLZKJ48R1S7H23Ji7oFO5Bw= @@ -169,8 +169,8 @@ golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= -golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= -golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= +golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= diff --git a/quic_connection.go b/quic_connection.go index 7db4899..f324245 100644 --- a/quic_connection.go +++ b/quic_connection.go @@ -54,6 +54,14 @@ func (s *QuicGoSendStream) CancelWrite(c uint64) { s.stream.CancelWrite(quic.StreamErrorCode(c)) } +func (s *QuicGoSendStream) SetPriority(p uint32) { + s.stream.SetPriority(p) +} + +func (s *QuicGoSendStream) SetIncremental(b bool) { + s.stream.SetIncremental(b) +} + type QUICGoConnection struct { conn *quic.Conn } diff --git a/send_flow.go b/send_flow.go index df4a066..4e8ca68 100644 --- a/send_flow.go +++ b/send_flow.go @@ -64,7 +64,7 @@ func (f *SendFlow) WriteRTPBytes(packet []byte) error { // NewSendStream creates a new Stream for sending outgoing RTP and RTCP packets // over a QUIC stream. -func (f *SendFlow) NewSendStream(ctx context.Context) (*RTPSendStream, error) { +func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremantal bool) (*RTPSendStream, error) { if err := f.isClosed(); err != nil { return nil, err } @@ -72,6 +72,9 @@ func (f *SendFlow) NewSendStream(ctx context.Context) (*RTPSendStream, error) { if err != nil { return nil, err } + s.SetPriority(priority) + s.SetIncremental(incremantal) + stream, err := newRTPSendStream(s, f.id, f.flowID, f.qlog) if err != nil { return nil, err diff --git a/session.go b/session.go index f17e2a7..062f836 100644 --- a/session.go +++ b/session.go @@ -19,6 +19,9 @@ type SendStream interface { io.Closer ID() int64 CancelWrite(uint64) + + SetPriority(p uint32) + SetIncremental(b bool) } type ReceiveStream interface { From 3f06af8a452e3853976aef16221a65de4b3d4e96 Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Tue, 14 Apr 2026 08:57:30 +0200 Subject: [PATCH 2/4] Allow priority if available, don't enforce it --- examples/interop/sender.go | 4 ++-- examples/playfromdisk/main.go | 2 +- go.mod | 2 -- go.sum | 4 ++-- integrationtests/integration_test.go | 2 +- quic_connection.go | 8 -------- send_flow.go | 9 +++++++-- session.go | 2 ++ 8 files changed, 15 insertions(+), 18 deletions(-) diff --git a/examples/interop/sender.go b/examples/interop/sender.go index 87fe1d7..3f1b71d 100644 --- a/examples/interop/sender.go +++ b/examples/interop/sender.go @@ -42,7 +42,7 @@ func (s *sender) send(flowID uint64, reader FrameReader, packetizer rtp.Packetiz } var singleStream *roq.RTPSendStream if sendMode == singleStreamMode { - singleStream, err = flow.NewSendStream(context.Background()) + singleStream, err = flow.NewSendStream(context.Background(), 0, false) if err != nil { return err } @@ -64,7 +64,7 @@ func (s *sender) send(flowID uint64, reader FrameReader, packetizer rtp.Packetiz return err } case streamPerFrameMode: - stream, err := flow.NewSendStream(context.Background()) + stream, err := flow.NewSendStream(context.Background(), 0, false) if err != nil { return err } diff --git a/examples/playfromdisk/main.go b/examples/playfromdisk/main.go index 57cdd1e..e2d8a26 100644 --- a/examples/playfromdisk/main.go +++ b/examples/playfromdisk/main.go @@ -104,7 +104,7 @@ func main() { panic(ivfErr) } - stream, err := flow.NewSendStream(context.Background()) + stream, err := flow.NewSendStream(context.Background(), 0, false) if err != nil { panic(err) } diff --git a/go.mod b/go.mod index 07192cc..2e4181e 100644 --- a/go.mod +++ b/go.mod @@ -23,5 +23,3 @@ require ( golang.org/x/time v0.14.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) - -replace github.com/quic-go/quic-go v0.57.1 => github.com/Willi-42/quic-go v0.7.1 diff --git a/go.sum b/go.sum index 40dbcb9..b4399ba 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,3 @@ -github.com/Willi-42/quic-go v0.7.1 h1:6Rjzjo1xmFbiGJ8Xib+CpB9imeJTEhmHs8jgGm1uceU= -github.com/Willi-42/quic-go v0.7.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -72,6 +70,8 @@ github.com/pion/webrtc/v3 v3.2.28 h1:ienStxZ6HcjtH2UlmnFpMM0loENiYjaX437uIUpQSKo github.com/pion/webrtc/v3 v3.2.28/go.mod h1:PNRCEuQlibrmuBhOTnol9j6KkIbUG11aHLEfNpUYey0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/quic-go v0.57.1 h1:25KAAR9QR8KZrCZRThWMKVAwGoiHIrNbT72ULHTuI10= +github.com/quic-go/quic-go v0.57.1/go.mod h1:ly4QBAjHA2VhdnxhojRsCUOeJwKYg+taDlos92xb1+s= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/sclevine/agouti v3.0.0+incompatible/go.mod h1:b4WX9W9L1sfQKXeJf1mUTLZKJ48R1S7H23Ji7oFO5Bw= diff --git a/integrationtests/integration_test.go b/integrationtests/integration_test.go index 0a9a139..ec4a7ce 100644 --- a/integrationtests/integration_test.go +++ b/integrationtests/integration_test.go @@ -48,7 +48,7 @@ func TestIntegration(t *testing.T) { sender := accept(t, ctx, l) f, err := sender.NewSendFlow(0) assert.NoError(t, err) - s, err := f.NewSendStream(ctx) + s, err := f.NewSendStream(ctx, 0, false) assert.NoError(t, err) for i := 0; i < 10; i++ { pkt := rtp.Packet{ diff --git a/quic_connection.go b/quic_connection.go index f324245..7db4899 100644 --- a/quic_connection.go +++ b/quic_connection.go @@ -54,14 +54,6 @@ func (s *QuicGoSendStream) CancelWrite(c uint64) { s.stream.CancelWrite(quic.StreamErrorCode(c)) } -func (s *QuicGoSendStream) SetPriority(p uint32) { - s.stream.SetPriority(p) -} - -func (s *QuicGoSendStream) SetIncremental(b bool) { - s.stream.SetIncremental(b) -} - type QUICGoConnection struct { conn *quic.Conn } diff --git a/send_flow.go b/send_flow.go index 4e8ca68..9a846e9 100644 --- a/send_flow.go +++ b/send_flow.go @@ -2,6 +2,7 @@ package roq import ( "context" + "log/slog" "sync" "github.com/mengelbart/qlog" @@ -72,8 +73,12 @@ func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremant if err != nil { return nil, err } - s.SetPriority(priority) - s.SetIncremental(incremantal) + priorityStream, ok := s.(PrioritySendStream) + if ok { + slog.Info("roq setting stream priority", "priority", priority, "incremental", incremantal) + priorityStream.SetPriority(priority) + priorityStream.SetIncremental(incremantal) + } stream, err := newRTPSendStream(s, f.id, f.flowID, f.qlog) if err != nil { diff --git a/session.go b/session.go index 062f836..81434df 100644 --- a/session.go +++ b/session.go @@ -19,7 +19,9 @@ type SendStream interface { io.Closer ID() int64 CancelWrite(uint64) +} +type PrioritySendStream interface { SetPriority(p uint32) SetIncremental(b bool) } From 9c8195816e462d42242e6fa63e2ab74a753f0165 Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Tue, 5 May 2026 11:42:11 +0200 Subject: [PATCH 3/4] Fix linter issues --- examples/interop/main.go | 12 +++++++----- examples/interop/receiver.go | 4 ++-- examples/interop/sender.go | 4 ++-- examples/playfromdisk/main.go | 2 +- examples/savetodisk/main.go | 4 ++-- session.go | 4 ++-- 6 files changed, 16 insertions(+), 14 deletions(-) diff --git a/examples/interop/main.go b/examples/interop/main.go index 33bbec9..b9d5a4a 100644 --- a/examples/interop/main.go +++ b/examples/interop/main.go @@ -88,7 +88,7 @@ func setupAndRun() error { if keyLog != nil { defer func() { log.Printf("closing keylog") - keyLog.Close() + keyLog.Close() //nolint }() } conn, err := connect(context.Background(), f, keyLog) @@ -142,7 +142,7 @@ func runSender(f flags, conn *quic.Conn) error { } qlogfile := getQLOGWriter("roq", role) if qlogfile != nil { - defer qlogfile.Close() + defer qlogfile.Close() //nolint } var qlogger *mqlog.Logger if qlogfile != nil { @@ -178,7 +178,9 @@ func runSender(f flags, conn *quic.Conn) error { return ffmpeg.Process.Kill() } time.Sleep(time.Second) - s.Close() + if err = s.Close(); err != nil { + return err + } return ffmpeg.Wait() } @@ -189,7 +191,7 @@ func runReceiver(f flags, conn *quic.Conn) error { } qlogfile := getQLOGWriter("roq", role) if qlogfile != nil { - defer qlogfile.Close() + defer qlogfile.Close() //nolint } var qlogger *mqlog.Logger if qlogfile != nil { @@ -221,7 +223,7 @@ func runReceiver(f flags, conn *quic.Conn) error { log.Println("WAITING FOR FFPLAY") //state, err := ffplay.Process.Wait() //log.Printf("ffmpeg returned %v, err: %v", state, err) - ffplay.Process.Kill() + ffplay.Process.Kill() //nolint }() writer, err = newIVFWriterWith(stdout, f.Codec) if err != nil { diff --git a/examples/interop/receiver.go b/examples/interop/receiver.go index f16e6de..3238f00 100644 --- a/examples/interop/receiver.go +++ b/examples/interop/receiver.go @@ -30,8 +30,8 @@ func (r *receiver) receive(flowID uint64, writer io.WriteCloser) error { if err != nil { return err } - defer flow.Close() - defer writer.Close() + defer flow.Close() //nolint + defer writer.Close() //nolint _, err = io.Copy(writer, flow) return err } diff --git a/examples/interop/sender.go b/examples/interop/sender.go index 3f1b71d..23ad792 100644 --- a/examples/interop/sender.go +++ b/examples/interop/sender.go @@ -46,9 +46,9 @@ func (s *sender) send(flowID uint64, reader FrameReader, packetizer rtp.Packetiz if err != nil { return err } - defer singleStream.Close() + defer singleStream.Close() //nolint } - defer flow.Close() + defer flow.Close() //nolint for { frame, err := reader.Read() if err != nil { diff --git a/examples/playfromdisk/main.go b/examples/playfromdisk/main.go index e2d8a26..3ff3232 100644 --- a/examples/playfromdisk/main.go +++ b/examples/playfromdisk/main.go @@ -120,6 +120,6 @@ func main() { panic(err) } } - stream.Close() + stream.Close() //nolint } } diff --git a/examples/savetodisk/main.go b/examples/savetodisk/main.go index 61c396a..81fbbd3 100644 --- a/examples/savetodisk/main.go +++ b/examples/savetodisk/main.go @@ -42,12 +42,12 @@ func main() { if err != nil { panic(err) } - defer flow.Close() + defer flow.Close() //nolint fileWriter, err := ivfwriter.New("output.ivf") if err != nil { panic(err) } - defer fileWriter.Close() + defer fileWriter.Close() //nolint for { buf := make([]byte, 4096) n, err := flow.Read(buf) diff --git a/session.go b/session.go index 81434df..c9fd931 100644 --- a/session.go +++ b/session.go @@ -149,8 +149,8 @@ func (s *Session) NewReceiveFlow(id uint64) (*ReceiveFlow, error) { func (s *Session) close(code uint64, reason string) { s.mutex.Lock() defer s.mutex.Unlock() - s.sendFlows.rangeFn(func(_ uint64, v *SendFlow) { v.Close() }) - s.receiveFlows.rangeFn(func(_ uint64, v *ReceiveFlow) { v.Close() }) + s.sendFlows.rangeFn(func(_ uint64, v *SendFlow) { _ = v.Close() }) + s.receiveFlows.rangeFn(func(_ uint64, v *ReceiveFlow) { _ = v.Close() }) _ = s.conn.CloseWithError(code, reason) s.closedErr = SessionError{ code: code, From 1736fbb402873dbacf956760588b08e4e54d2322 Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Tue, 5 May 2026 11:42:23 +0200 Subject: [PATCH 4/4] Remove unused constant --- examples/interop/receiver.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/examples/interop/receiver.go b/examples/interop/receiver.go index 3238f00..5cb4090 100644 --- a/examples/interop/receiver.go +++ b/examples/interop/receiver.go @@ -7,10 +7,6 @@ import ( "github.com/mengelbart/roq" ) -const ( - bufferSize = 1 << 15 -) - type receiver struct { session *roq.Session }