1package capture
2
3import (
4 "cmp"
5 "context"
6 "errors"
7 "fmt"
8 "sync"
9 "time"
10
11 "github.com/bnema/neferwl/internal/adapters/syncfile"
12 "github.com/bnema/neferwl/internal/ports"
13)
14
15// MaxRequests bounds requests collected for one output frame.
16const MaxRequests = 8
17
18const pipelineSlots = 2
19
20// ErrSecurityState rejects capture admission across an epoch or while protected.
21var ErrSecurityState = errors.New("capture security epoch changed or session protected")
22
23// batch remains owned by the worker until it is returned through completed.
24// Both batches and their request storage are reused for the output's lifetime.
25type batch struct {
26 frame ports.CaptureFrame
27 requests [MaxRequests]ports.CaptureRequest
28 count int
29 at time.Time
30 // gated: the worker delivers the requests only when the owner releases
31 // the batch (see Pipeline.BeginGate). release carries its verdict; it is
32 // made once and lives as long as the batch.
33 gated bool
34 release chan error
35}
36
37// Pipeline moves fence waiting and SHM copies off the renderer's owner.
38// Submit, Recycle and Close belong to that owner; one worker handles all jobs.
39type Pipeline struct {
40 // Set before owner use; no mutation or worker access to this dependency.
41 Security ports.SessionSecurity
42 ctx context.Context
43 cancel context.CancelFunc
44 replies chan<- ports.CaptureDone
45 jobs chan *batch
46 completed chan *batch
47 batches [pipelineSlots]batch
48 // gating is set between BeginGate and EndGate; held are the batches
49 // submitted meanwhile, waiting for their verdict. Owner only.
50 gating bool
51 held [pipelineSlots]*batch
52 nheld int
53 excl excluder // scratch of ExcludedScene, owner only
54 off *offscreen
55 worker sync.WaitGroup
56}
57
58func NewPipeline(ctx context.Context, replies chan<- ports.CaptureDone) *Pipeline {
59 ctx, cancel := context.WithCancel(ctx)
60 p := &Pipeline{ctx: ctx, cancel: cancel, replies: replies, jobs: make(chan *batch, pipelineSlots), completed: make(chan *batch, pipelineSlots)}
61 for i := range p.batches {
62 p.batches[i].release = make(chan error, 1)
63 }
64 p.worker.Go(p.run)
65 return p
66}
67
68// Completed wakes the owner even when the output is otherwise idle. Its opaque
69// token is passed directly to Recycle; callers never access batch storage.
70func (p *Pipeline) Completed() <-chan *batch { return p.completed }
71
72// Recycle returns a completed GPU readback lease on the renderer's owner.
73func (p *Pipeline) Recycle(b *batch, r ports.Renderer) {
74 r.EndCapture(b.frame)
75 rel := b.release
76 select { // a verdict nobody took (the worker was cancelled) must not leak to the next use
77 case <-rel:
78 default:
79 }
80 *b = batch{release: rel}
81}
82
83// GateVerdict is what a batch held by the indicator gate is failed with when
84// the frame that carries the indicator did not reach the display: err wrapped
85// in ErrIndicatorMissing; nil stays nil (released).
86func GateVerdict(err error) error {
87 if err == nil {
88 return nil
89 }
90 return fmt.Errorf("%w: %w", ErrIndicatorMissing, err)
91}
92
93// ErrFrameNotPresented fails a gated batch whose owner never gave a verdict:
94// the frame it waited for was dropped on a path that did not present it.
95var ErrFrameNotPresented = fmt.Errorf("%w: frame not presented", ErrIndicatorMissing)
96
97// BeginGate makes the batches submitted from now on wait for EndGate: their
98// worker does its fence wait, then holds the pixels back until the owner says
99// the frame that carries the capture indicator is on screen. See EndGate.
100func (p *Pipeline) BeginGate() { p.gating = true }
101
102// EndGate gives the verdict on the batches held since BeginGate and stops
103// gating. A nil err releases them: their captures are written and answered. A
104// non-nil err fails every one of their requests with it: no pixel is
105// delivered. It covers the child pipeline of a hidden workspace too. Nothing
106// held, nothing done: no allocation, no cost while no capture is gated.
107func (p *Pipeline) EndGate(err error) {
108 if p == nil {
109 return
110 }
111 p.gating = false
112 for i := range p.nheld {
113 b := p.held[i]
114 p.held[i] = nil
115 select {
116 case b.release <- err:
117 default: // the verdict is already there
118 }
119 }
120 p.nheld = 0
121 if p.off != nil && p.off.p != nil {
122 p.off.p.EndGate(err)
123 }
124}
125
126// SubmitScoped consumes requests with explicit immutable scene admission. The
127// final gate read precedes BeginCapture, not the render that produced pixels.
128func (p *Pipeline) SubmitScoped(state ports.SecurityState, r ports.Renderer, requests []ports.CaptureRequest) {
129 if len(requests) == 0 {
130 return
131 }
132 // Reclaim returned leases before declaring the small ring saturated.
133 p.reclaim(r)
134 var b *batch
135 if len(requests) <= MaxRequests && p.ctx.Err() == nil {
136 for i := range p.batches {
137 if p.batches[i].frame == nil {
138 b = &p.batches[i]
139 break
140 }
141 }
142 }
143 var err error
144 if b == nil {
145 err = fmt.Errorf("capture pipeline full or stopped")
146 } else if !p.captureAllowed(state) {
147 err = ErrSecurityState
148 } else {
149 b.frame, err = r.BeginCapture()
150 if err == nil && b.frame == nil {
151 err = fmt.Errorf("renderer returned no capture frame")
152 }
153 }
154 if err != nil {
155 for _, q := range requests {
156 Fail(p.ctx, q, err, p.replies)
157 }
158 return
159 }
160 b.count = copy(b.requests[:], requests)
161 b.at = monotonicNow()
162 if b.gated = p.gating; b.gated {
163 p.held[p.nheld] = b
164 p.nheld++
165 }
166 // There are exactly two reusable batches and two queue slots, so this
167 // send cannot wait: no batch can be queued twice before recycling.
168 p.jobs <- b
169}
170
171func (p *Pipeline) captureAllowed(state ports.SecurityState) bool {
172 return p.Security == nil || !state.Protected && p.Security.Snapshot() == state
173}
174
175func (p *Pipeline) reclaim(r ports.Renderer) {
176 for {
177 select {
178 case b := <-p.completed:
179 p.Recycle(b, r)
180 default:
181 return
182 }
183 }
184}
185
186func (p *Pipeline) run() {
187 for {
188 select {
189 case b := <-p.jobs:
190 p.process(b)
191 case <-p.ctx.Done():
192 for {
193 select {
194 case b := <-p.jobs:
195 p.process(b)
196 default:
197 return
198 }
199 }
200 }
201 }
202}
203
204func (p *Pipeline) process(b *batch) {
205 err := p.ctx.Err()
206 if err == nil && b.frame.Done() != nil {
207 err = syncfile.Wait(p.ctx, b.frame.Done())
208 }
209 if b.gated {
210 // The verdict is always taken, so none is left in the channel. The
211 // owner's verdict, when it gave one, says why the capture failed
212 // better than a cancelled context.
213 var v error
214 select {
215 case v = <-b.release:
216 case <-p.ctx.Done():
217 select {
218 case v = <-b.release:
219 default:
220 }
221 }
222 err = cmp.Or(v, err, p.ctx.Err())
223 }
224 for i := range b.count {
225 q := b.requests[i]
226 if err != nil {
227 Fail(p.ctx, q, err, p.replies)
228 } else if e := p.ctx.Err(); e != nil {
229 Fail(p.ctx, q, e, p.replies)
230 } else {
231 writeFrame(p.ctx, q, b.frame, b.at, p.replies)
232 }
233 }
234 p.completed <- b
235}
236
237// Close joins the worker before any renderer memory may be unmapped. A
238// cancelled GPU operation is returned without reading its staging buffer;
239// EndCapture must retain GPU-pending storage until completion.
240func (p *Pipeline) Close(r ports.Renderer) {
241 p.cancel()
242 p.worker.Wait()
243 p.reclaim(r)
244 if p.off != nil {
245 p.off.close()
246 }
247}