1package core
2
3import (
4 "errors"
5 "fmt"
6 "log/slog"
7 "math"
8 "strings"
9 "sync"
10 "sync/atomic"
11 "time"
12 "unsafe"
13
14 "github.com/bnema/purego"
15
16 portin "github.com/bnema/purego-pipewire/internal/ports/in"
17 portout "github.com/bnema/purego-pipewire/internal/ports/out"
18 "github.com/bnema/purego-pipewire/spa"
19)
20
21// Video source limits. They are fixed so that nothing is allocated after
22// NewVideoSource except the memfd mappings made in add_buffer.
23const (
24 videoMaxBuffers = 16 // Buffers param maximum, size of the buffer slot table
25 videoMaxSources = 64 // size of the process-wide callback slot table
26 videoPodSize = 512
27
28 pwStreamFlagDriver = 0x8
29 pwStreamFlagAllocBuffers = 0x100
30 pwStreamStateStreaming = 3
31 pwStreamStatePaused = 2
32
33 defaultVideoFPS = 60
34)
35
36var (
37 ErrVideoConfig = errors.New("invalid video source config")
38 ErrVideoTableFull = errors.New("too many video sources")
39 ErrVideoStreamCreate = errors.New("failed to create video stream")
40 ErrVideoConnect = errors.New("failed to connect video stream")
41 ErrVideoResize = errors.New("failed to resize video stream")
42 ErrVideoBorrowed = errors.New("video buffer still borrowed")
43)
44
45// VideoConfig configures a VideoSource. MaxFPS defaults to 60 and Formats to
46// BGRx. Only BGRx and BGRA are supported: buffers hold 4 bytes per pixel.
47type VideoConfig struct {
48 Name string
49 Width, Height int
50 MaxFPS int
51 Formats []spa.VideoFormat
52}
53
54// VideoBuffer is a buffer that was handed out by VideoSource.Dequeue. Data
55// aliases a memfd mapping owned by the source: it stays valid until the buffer
56// is given back with Queue or Release, and not afterwards. The structure is
57// owned by the source; do not copy it, retain it or modify it.
58type VideoBuffer struct {
59 Index int
60 Data []byte
61 Stride, Width, Height int
62 Format spa.VideoFormat
63}
64
65// C layouts (amd64/arm64). Offsets are asserted in tests.
66
67// videoPWBuffer mirrors struct pw_buffer; UserData carries slot index+1.
68type videoPWBuffer struct {
69 Buffer *spaBuffer
70 UserData uintptr
71 Size uint64
72}
73
74// spaMeta mirrors struct spa_meta.
75type spaMeta struct {
76 Type uint32
77 Size uint32
78 Data unsafe.Pointer
79}
80
81// spaMetaHeader mirrors struct spa_meta_header.
82type spaMetaHeader struct {
83 Flags uint32
84 Offset uint32
85 PTS int64
86 DTSOffset int64
87 Seq uint64
88}
89
90// videoEvents mirrors struct pw_stream_events (version 2).
91type videoEvents struct {
92 Version uint32
93 Destroy uintptr
94 StateChanged uintptr
95 ControlInfo uintptr
96 IOChanged uintptr
97 ParamChanged uintptr
98 AddBuffer uintptr
99 RemoveBuf uintptr
100 Process uintptr
101 Drained uintptr
102 Command uintptr
103 TriggerDone uintptr
104}
105
106type videoSlot struct {
107 buf VideoBuffer
108 pw unsafe.Pointer // struct pw_buffer; nil once removed
109 hdr *spaMetaHeader // nil if the buffer has no header meta
110 chunk *spaChunk
111 fd int
112 mem []byte
113 live bool // slot holds a mapped buffer
114 removed bool // PipeWire removed the buffer while it was borrowed
115 borrowed bool
116}
117
118// VideoSource is a PipeWire Video/Source node that drives the graph and
119// exposes memfd-backed buffers. Every method, and every handler callback, runs
120// on the loop thread: the goroutine (locked to its OS thread) that calls
121// Loop.Iterate.
122//
123// Frames are produced with Dequeue, then writing Data, then Queue, then
124// Trigger. The buffer must be queued before Trigger, because the process
125// callback does nothing: PipeWire picks up queued buffers when triggered.
126type VideoSource struct {
127 capi portout.VideoStreamAPI
128 shm portout.SharedMemAPI
129 handler portin.VideoHandler
130
131 stream unsafe.Pointer
132 table int // index in videoTable
133 events videoEvents
134 hook [8]uint64 // struct spa_hook, 48 bytes
135 params [2]unsafe.Pointer
136 enumPod [videoPodSize]byte
137 buffPod [videoPodSize]byte
138 metaPod [128]byte
139 cfg VideoConfig
140 formats [4]spa.VideoFormat
141 nformats int
142
143 state int32
144 nodeID uint32
145 format spa.VideoFormat
146 width int
147 height int
148 stride int
149 negotiated bool
150 closed bool
151 warned bool
152
153 slots [videoMaxBuffers]videoSlot
154}
155
156var (
157 videoTable [videoMaxSources]atomic.Pointer[VideoSource]
158
159 videoCBOnce sync.Once
160 videoCB struct {
161 stateChanged, paramChanged, addBuffer, removeBuffer, process, triggerDone uintptr
162 }
163)
164
165// uptr turns a raw callback argument into a pointer without a
166// uintptr-to-unsafe.Pointer conversion; the value is C memory or a small
167// integer, never Go memory.
168func uptr(v uintptr) unsafe.Pointer { return *(*unsafe.Pointer)(unsafe.Pointer(&v)) }
169
170func videoLookup(a *purego.CallbackArgs) *VideoSource {
171 idx := int(a.Int(0)) - 1
172 if uint(idx) >= videoMaxSources {
173 return nil
174 }
175 return videoTable[idx].Load()
176}
177
178func registerVideoCallbacks() {
179 videoCBOnce.Do(func() {
180 videoCB.stateChanged = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
181 if s := videoLookup(a); s != nil {
182 s.onStateChanged(int32(a.Int(2)), uptr(a.Int(3)))
183 }
184 return 0
185 })
186 videoCB.paramChanged = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
187 if s := videoLookup(a); s != nil {
188 s.onParamChanged(uint32(a.Int(1)), uptr(a.Int(2)))
189 }
190 return 0
191 })
192 videoCB.addBuffer = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
193 if s := videoLookup(a); s != nil {
194 s.onAddBuffer(uptr(a.Int(1)))
195 }
196 return 0
197 })
198 videoCB.removeBuffer = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
199 if s := videoLookup(a); s != nil {
200 s.onRemoveBuffer(uptr(a.Int(1)))
201 }
202 return 0
203 })
204 // process and trigger_done intentionally do nothing: buffers are
205 // queued by the owner before Trigger.
206 videoCB.process = purego.NewCallbackInts(func(*purego.CallbackArgs) uintptr { return 0 })
207 videoCB.triggerDone = purego.NewCallbackInts(func(*purego.CallbackArgs) uintptr { return 0 })
208 })
209}
210
211// NewVideoSource creates and connects a Video/Source stream on the core. The
212// stream drives the graph (DRIVER) and allocates its own memfd buffers
213// (ALLOC_BUFFERS). Call it, and every other method, on the loop thread.
214func (c *Core) NewVideoSource(cfg VideoConfig, h portin.VideoHandler) (*VideoSource, error) {
215 if c.ptr == nil || c.capi == nil {
216 return nil, ErrVideoStreamCreate
217 }
218 if h == nil || !validVideoSize(cfg.Width, cfg.Height) {
219 return nil, ErrVideoConfig
220 }
221 if cfg.MaxFPS <= 0 {
222 cfg.MaxFPS = defaultVideoFPS
223 }
224 s := &VideoSource{capi: c.capi, shm: c.shm, handler: h, cfg: cfg, table: -1}
225 if len(cfg.Formats) == 0 {
226 s.formats[0], s.nformats = spa.VideoFormatBGRx, 1
227 } else {
228 if len(cfg.Formats) > len(s.formats) {
229 return nil, ErrVideoConfig
230 }
231 for _, f := range cfg.Formats {
232 if f != spa.VideoFormatBGRx && f != spa.VideoFormatBGRA {
233 return nil, fmt.Errorf("%w: unsupported format %d", ErrVideoConfig, f)
234 }
235 }
236 s.nformats = copy(s.formats[:], cfg.Formats)
237 }
238 s.cfg.Formats = nil // the slot-owned array is the source of truth
239 name := cfg.Name
240 if name == "" {
241 name = "purego-pipewire"
242 }
243 if strings.ContainsAny(name, "\"\\") || strings.ContainsFunc(name, func(r rune) bool { return r < 0x20 || r == 0x7f }) {
244 return nil, fmt.Errorf("%w: invalid name %q", ErrVideoConfig, name)
245 }
246 if _, err := spa.VideoEnumFormat(s.enumPod[:], cfg.Width, cfg.Height, cfg.MaxFPS, s.formats[:s.nformats]...); err != nil {
247 return nil, fmt.Errorf("%w: %v", ErrVideoConfig, err)
248 }
249
250 registerVideoCallbacks()
251 for i := range videoTable {
252 if videoTable[i].CompareAndSwap(nil, s) {
253 s.table = i
254 break
255 }
256 }
257 if s.table < 0 {
258 return nil, ErrVideoTableFull
259 }
260
261 cname := append([]byte(name), 0)
262 props := append([]byte(`media.class=Video/Source node.name="`+name+`"`), 0)
263 propsPtr := c.capi.PWPropertiesNewString(&props[0])
264 if propsPtr == nil {
265 videoTable[s.table].Store(nil)
266 return nil, ErrVideoStreamCreate
267 }
268 s.stream = c.capi.PWStreamNew(c.ptr, &cname[0], propsPtr) // takes ownership of props
269 if s.stream == nil {
270 videoTable[s.table].Store(nil)
271 return nil, ErrVideoStreamCreate
272 }
273
274 s.events = videoEvents{
275 Version: 2,
276 StateChanged: videoCB.stateChanged,
277 ParamChanged: videoCB.paramChanged,
278 AddBuffer: videoCB.addBuffer,
279 RemoveBuf: videoCB.removeBuffer,
280 Process: videoCB.process,
281 TriggerDone: videoCB.triggerDone,
282 }
283 data := uptr(uintptr(s.table + 1))
284 c.capi.PWStreamAddListener(s.stream, unsafe.Pointer(&s.hook[0]), unsafe.Pointer(&s.events), data)
285
286 s.params[0] = unsafe.Pointer(&s.enumPod[0])
287 ret := c.capi.PWStreamConnect(s.stream, 1 /* PW_DIRECTION_OUTPUT */, spa.IDInvalid,
288 pwStreamFlagDriver|pwStreamFlagAllocBuffers, unsafe.Pointer(&s.params[0]), 1)
289 if ret < 0 {
290 s.closed = true
291 s.teardown()
292 return nil, fmt.Errorf("%w: pw_stream_connect returned %d", ErrVideoConnect, ret)
293 }
294 return s, nil
295}
296
297// NodeID returns the PipeWire node id, or 0 until the stream is paused or
298// streaming.
299func (s *VideoSource) NodeID() uint32 { return s.nodeID }
300
301// State returns the last stream state (see VideoHandler.StateChanged).
302func (s *VideoSource) State() int32 { return s.state }
303
304// Dequeue returns a free buffer to draw into, or nil when none is available.
305// The buffer is borrowed until Queue or Release. Buffers that could not be
306// given a memfd are skipped (they stay out of the queue).
307func (s *VideoSource) Dequeue() *VideoBuffer {
308 if s.stream == nil || !s.negotiated {
309 return nil
310 }
311 for i := 0; i < 2*videoMaxBuffers; i++ {
312 p := s.capi.PWStreamDequeueBuffer(s.stream)
313 if p == nil {
314 return nil
315 }
316 idx := int((*videoPWBuffer)(p).UserData) - 1
317 if uint(idx) >= videoMaxBuffers {
318 continue // unusable buffer: never handed out
319 }
320 sl := &s.slots[idx]
321 if !sl.live || sl.removed || sl.borrowed || sl.pw != p {
322 continue
323 }
324 sl.borrowed = true
325 return &sl.buf
326 }
327 return nil
328}
329
330// slotOf validates b against the slot table.
331func (s *VideoSource) slotOf(b *VideoBuffer) *videoSlot {
332 if b == nil || uint(b.Index) >= videoMaxBuffers {
333 return nil
334 }
335 sl := &s.slots[b.Index]
336 if &sl.buf != b || !sl.live || !sl.borrowed {
337 return nil
338 }
339 return sl
340}
341
342// Queue hands a filled buffer back to PipeWire with its frame sequence and
343// presentation time. When corrupted is set the frame is flagged corrupted and
344// carries no data. Call Trigger after Queue: the buffer is queued before the
345// graph cycle is triggered. Queue on a buffer PipeWire already removed only
346// releases it.
347func (s *VideoSource) Queue(b *VideoBuffer, seq uint64, pts time.Duration, corrupted bool) {
348 sl := s.slotOf(b)
349 if sl == nil {
350 return
351 }
352 if sl.removed || s.stream == nil {
353 s.freeSlot(sl)
354 return
355 }
356 size := uint32(sl.buf.Stride * sl.buf.Height)
357 if c := sl.chunk; c != nil {
358 c.Offset, c.Size, c.Stride, c.Flags = 0, size, int32(sl.buf.Stride), 0
359 if corrupted {
360 c.Size, c.Flags = 0, spa.ChunkFlagCorrupted
361 }
362 }
363 if h := sl.hdr; h != nil {
364 h.Flags, h.Offset, h.PTS, h.DTSOffset, h.Seq = 0, 0, int64(pts), 0, seq
365 if corrupted {
366 h.Flags = spa.MetaHeaderFlagCorrupted
367 }
368 }
369 sl.borrowed = false
370 s.capi.PWStreamQueueBuffer(s.stream, sl.pw)
371}
372
373// Release gives a dequeued buffer back without queueing a frame. If PipeWire
374// removed the buffer in the meantime its memory is unmapped now. b.Data must
375// not be used afterwards.
376func (s *VideoSource) Release(b *VideoBuffer) {
377 sl := s.slotOf(b)
378 if sl == nil {
379 return
380 }
381 if sl.removed || s.stream == nil {
382 s.freeSlot(sl)
383 return
384 }
385 sl.borrowed = false
386 s.capi.PWStreamReturnBuffer(s.stream, sl.pw)
387}
388
389// Trigger starts a graph cycle (pw_stream_trigger_process). Queue the filled
390// buffer first.
391func (s *VideoSource) Trigger() {
392 if s.stream != nil {
393 s.capi.PWStreamTriggerProcess(s.stream)
394 }
395}
396
397// validVideoSize reports whether a w×h frame of 4-byte pixels fits the
398// int32 sizes SPA buffers carry.
399func validVideoSize(w, h int) bool {
400 return w > 0 && h > 0 && w <= 1<<15 && h <= 1<<15 && w*4*h <= math.MaxInt32
401}
402
403// Resize renegotiates the stream with a new size. It fails with
404// ErrVideoBorrowed while a buffer is borrowed: PipeWire replaces every
405// buffer, and the replacements need the slots the borrowed ones hold.
406func (s *VideoSource) Resize(w, h int) error {
407 if s.stream == nil {
408 return ErrVideoResize
409 }
410 if !validVideoSize(w, h) {
411 return ErrVideoConfig
412 }
413 for i := range s.slots {
414 if s.slots[i].borrowed {
415 return ErrVideoBorrowed
416 }
417 }
418 if _, err := spa.VideoEnumFormat(s.enumPod[:], w, h, s.cfg.MaxFPS, s.formats[:s.nformats]...); err != nil {
419 return fmt.Errorf("%w: %v", ErrVideoResize, err)
420 }
421 s.cfg.Width, s.cfg.Height = w, h
422 s.params[0] = unsafe.Pointer(&s.enumPod[0])
423 if ret := s.capi.PWStreamUpdateParams(s.stream, unsafe.Pointer(&s.params[0]), 1); ret < 0 {
424 return fmt.Errorf("%w: pw_stream_update_params returned %d", ErrVideoResize, ret)
425 }
426 return nil
427}
428
429// Destroy disconnects and destroys the stream. PipeWire removes every buffer
430// during the disconnect; buffers that are still borrowed keep their mapping
431// until Release (or Queue). Call Destroy only when no buffer is borrowed,
432// unless the caller releases them afterwards. No handler callback runs after
433// Destroy starts. Destroy is idempotent.
434func (s *VideoSource) Destroy() {
435 s.closed = true
436 s.teardown()
437}
438
439func (s *VideoSource) teardown() {
440 if s.stream != nil {
441 stream := s.stream
442 s.capi.PWStreamDisconnect(stream)
443 s.capi.PWStreamDestroy(stream)
444 s.stream = nil
445 }
446 if s.table >= 0 {
447 videoTable[s.table].Store(nil)
448 s.table = -1
449 }
450 for i := range s.slots {
451 sl := &s.slots[i]
452 if !sl.live {
453 continue
454 }
455 sl.pw, sl.hdr, sl.chunk = nil, nil, nil
456 if sl.borrowed {
457 sl.removed = true
458 } else {
459 s.freeSlot(sl)
460 }
461 }
462}
463
464// freeSlot unmaps the slot's memory and makes the slot reusable.
465func (s *VideoSource) freeSlot(sl *videoSlot) {
466 if sl.live {
467 s.shm.FreeSharedMem(sl.fd, sl.mem)
468 }
469 *sl = videoSlot{}
470}
471
472// --- stream event handlers (loop thread) ---
473
474func (s *VideoSource) onStateChanged(state int32, errp unsafe.Pointer) {
475 s.state = state
476 if state == pwStreamStatePaused || state == pwStreamStateStreaming {
477 s.nodeID = s.capi.PWStreamGetNodeId(s.stream)
478 }
479 if state == pwStreamStateStreaming && !s.warned {
480 s.warned = true
481 if !s.capi.PWStreamIsDriving(s.stream) {
482 slog.Warn("purego-pipewire: video stream is not driving the graph", "node", s.nodeID)
483 }
484 }
485 if s.closed {
486 return
487 }
488 msg := ""
489 if state < 0 {
490 msg = cString(errp)
491 }
492 s.handler.StateChanged(state, s.nodeID, msg)
493}
494
495func cString(p unsafe.Pointer) string {
496 if p == nil {
497 return ""
498 }
499 n := 0
500 for n < 512 && *(*byte)(unsafe.Add(p, n)) != 0 {
501 n++
502 }
503 return string(unsafe.Slice((*byte)(p), n))
504}
505
506func (s *VideoSource) onParamChanged(id uint32, param unsafe.Pointer) {
507 if id != spa.ParamFormat || s.stream == nil {
508 return
509 }
510 if param == nil {
511 s.negotiated = false
512 return
513 }
514 size := int(*(*uint32)(param))
515 if size < 8 || size > 4096 {
516 return
517 }
518 pod := unsafe.Slice((*byte)(param), 8+size)
519 f, w, h, ok := spa.ParseVideoFormat(pod)
520 if !ok {
521 return
522 }
523 format := spa.VideoFormat(f)
524 allowed := false
525 for _, af := range s.formats[:s.nformats] {
526 allowed = allowed || af == format
527 }
528 if !allowed {
529 return
530 }
531 stride := w * 4
532 buffers, err := spa.VideoBuffers(s.buffPod[:], stride*h, stride, 2, 8, videoMaxBuffers, 1<<spa.DataMemFd)
533 if err != nil {
534 return
535 }
536 meta, err := spa.MetaHeader(s.metaPod[:])
537 if err != nil {
538 return
539 }
540 s.format, s.width, s.height, s.stride, s.negotiated = format, w, h, stride, true
541 s.params[0] = unsafe.Pointer(&buffers[0])
542 s.params[1] = unsafe.Pointer(&meta[0])
543 s.capi.PWStreamUpdateParams(s.stream, unsafe.Pointer(&s.params[0]), 2)
544 if !s.closed {
545 s.handler.FormatChanged(format, w, h, stride)
546 }
547}
548
549func (s *VideoSource) onAddBuffer(p unsafe.Pointer) {
550 if p == nil {
551 return
552 }
553 pb := (*videoPWBuffer)(p)
554 pb.UserData = 0 // unusable until a slot is assigned
555 sb := pb.Buffer
556 if sb == nil || sb.NDatas == 0 || sb.Datas == nil || !s.negotiated {
557 return
558 }
559 d := (*spaData)(sb.Datas)
560 if d.Type&(1<<spa.DataMemFd) == 0 {
561 return // DMA-BUF only: not supported
562 }
563 idx := -1
564 for i := range s.slots {
565 if !s.slots[i].live {
566 idx = i
567 break
568 }
569 }
570 if idx < 0 {
571 return
572 }
573 size := s.stride * s.height
574 fd, mem, err := s.shm.NewSharedMem("pw-video", size)
575 if err != nil || len(mem) < size {
576 if err == nil {
577 s.shm.FreeSharedMem(fd, mem)
578 }
579 return
580 }
581 d.Type, d.Flags, d.Fd = spa.DataMemFd, spa.DataFlagReadWrite, int64(fd)
582 d.MapOffset, d.Maxsize, d.Data = 0, uint32(size), unsafe.Pointer(&mem[0])
583 if d.Chunk != nil {
584 d.Chunk.Offset, d.Chunk.Size, d.Chunk.Stride, d.Chunk.Flags = 0, uint32(size), int32(s.stride), 0
585 }
586 sl := &s.slots[idx]
587 *sl = videoSlot{
588 buf: VideoBuffer{Index: idx, Data: mem, Stride: s.stride, Width: s.width, Height: s.height, Format: s.format},
589 pw: p,
590 chunk: d.Chunk,
591 fd: fd,
592 mem: mem,
593 live: true,
594 }
595 if sb.Metas != nil {
596 metas := unsafe.Slice((*spaMeta)(sb.Metas), int(sb.NMetas))
597 for i := range metas {
598 if metas[i].Type == spa.MetaTypeHeader && metas[i].Size >= spa.MetaHeaderSize && metas[i].Data != nil {
599 sl.hdr = (*spaMetaHeader)(metas[i].Data)
600 break
601 }
602 }
603 }
604 pb.UserData = uintptr(idx + 1)
605 if !s.closed {
606 s.handler.BuffersChanged()
607 }
608}
609
610func (s *VideoSource) onRemoveBuffer(p unsafe.Pointer) {
611 if p == nil {
612 return
613 }
614 pb := (*videoPWBuffer)(p)
615 idx := int(pb.UserData) - 1
616 if uint(idx) >= videoMaxBuffers {
617 return
618 }
619 sl := &s.slots[idx]
620 if !sl.live || sl.pw != p {
621 return
622 }
623 pb.UserData = 0
624 if sl.borrowed {
625 // The owner may still write into the mapping: keep it until Release.
626 sl.removed = true
627 sl.pw, sl.hdr, sl.chunk = nil, nil, nil
628 } else {
629 s.freeSlot(sl)
630 }
631 if !s.closed {
632 s.handler.BuffersChanged()
633 }
634}