forgectrl: streams preempt - the newest camera request wins the mux

A stream request for the other camera kicks current stream clients via a generation counter (their streams end cleanly; viewers freeze on the last frame) and switches. The index page retry consults /cam/status first so a preempted view does not steal the camera back.
This commit is contained in:
ScottW514
2026-08-03 12:47:16 -04:00
parent 247cfacb40
commit dc34fc6602
4 changed files with 55 additions and 35 deletions
+17 -15
View File
@@ -142,26 +142,28 @@ Engine model: one worker owns the V4L2 node persistently (media-ctl /
v4l2-ctl configure sequences identical to gfhardware/cam.py, factory v4l2-ctl configure sequences identical to gfhardware/cam.py, factory
exposure/gain/WB, software hflip in the demosaic); starts on demand, exposure/gain/WB, software hflip in the demosaic); starts on demand,
full teardown after 10 s idle so gfhardware one-shot grabs still work. full teardown after 10 s idle so gfhardware one-shot grabs still work.
The cameras share the hardware video-mux; arbitration: The cameras share the hardware video-mux; the NEWEST request wins it
- **Stream clients pin the selection.** A STREAM request for the other (single-operator model):
camera waits up to 3 s for the pin to drain (absorbs page - **Streams preempt.** A STREAM request for the other camera kicks the
navigations), then returns 409. The index page therefore runs a current stream clients - their streams end cleanly (viewers freeze on
single toggled stream, never two. the last frame) - and switches. The only stream failure mode is a
- **Snapshots never fail busy.** A snapshot of the other camera makes switch timeout (a kicked client not draining within 3 s).
the worker *borrow* the mux: pause the stream, switch, grab one - **Snapshots borrow.** A snapshot of the other camera does not switch:
frame, switch back (~1-2 s freeze for stream viewers; "Head peek" on the worker pauses the stream, switches, grabs one frame, switches
the index page uses this). Arbitration compares against the engine's back (~1-2 s freeze; "Head peek" on the index page uses this).
home camera, so requests racing the borrow window still 409. Arbitration compares against the engine's home camera, so stream
requests racing the borrow window preempt correctly.
The per-camera lamp (`pic/lid_led` / `head/white_led`) is raised to The per-camera lamp (`pic/lid_led` / `head/white_led`) is raised to
`FORGECTRL_LAMP` (default 132) while capturing and restored on idle. `FORGECTRL_LAMP` (default 132) while capturing and restored on idle.
Bench (2026-08-03, on the board): stream 3.2 fps sustained at 1296×972; Bench (2026-08-03, on the board): stream 3.2 fps sustained at 1296×972;
full-res snapshot 2.4 s warm / 2.7 s cold (cold includes the pipeline full-res snapshot 2.4 s warm / 2.7 s cold (cold includes the pipeline
bring-up); camera switch works both ways; two parallel clients share the bring-up); two parallel same-camera clients share the frame rate; idle
frame rate; idle teardown observed. Borrow verified: head snapshot 200 teardown observed. Borrow verified: head snapshot 200 during a lid
during a lid stream, the stream rode through the gap (120 frames over a stream, the stream riding through the ~1-2 s gap. Preemption verified:
40 s window with one borrow), and head-stream requests fired both into a head-stream request ended the lid viewer's stream cleanly (curl exit
the borrow window and during normal streaming got 409. **Motion coexistence proven**: X 0 mid-stream) and was serving head frames within ~2 s; switching back
likewise. **Motion coexistence proven**: X
round-trip jogs at F1200 with an active stream — producer stats round-trip jogs at F1200 with an active stream — producer stats
`clamped 0`, max behind 4.5 ms (the daemon runs at nice +5, single `clamped 0`, max behind 4.5 ms (the daemon runs at nice +5, single
core). Run by hand: `/usr/bin/forgectrl >> /data/forgectrl.log 2>&1 &` core). Run by hand: `/usr/bin/forgectrl >> /data/forgectrl.log 2>&1 &`
@@ -91,6 +91,8 @@ static struct {
cam_id_t home_cam; /* camera the engine serves for streaming - cam_id_t home_cam; /* camera the engine serves for streaming -
* what arbitration must compare against */ * what arbitration must compare against */
int clients; int clients;
uint64_t kick_gen; /* bumped to preempt all current stream
* clients (their streams end cleanly) */
struct timespec last_activity; struct timespec last_activity;
/* published stream frame (half-res JPEG) */ /* published stream frame (half-res JPEG) */
@@ -727,9 +729,15 @@ static int ensure_engine(cam_id_t cam, char *err, size_t errlen)
if (running && cur != cam) { if (running && cur != cam) {
if (clients > 0) { if (clients > 0) {
/* Grace: a client that just disconnected releases its pin /* Last request wins: preempt the current stream clients
* only when the MHD send fails on the next frame - absorb * (single-operator machine - the newest ask is the
* that (page navigations) instead of failing instantly. */ * operator). Kicked clients wake, end their streams
* cleanly (viewers freeze on their last frame), and
* release their pins; wait for that to drain. */
pthread_mutex_lock(&eng.lock);
eng.kick_gen++;
pthread_cond_broadcast(&eng.frame_cv);
pthread_mutex_unlock(&eng.lock);
struct timespec t0, t; struct timespec t0, t;
now_ts(&t0); now_ts(&t0);
do { do {
@@ -741,8 +749,8 @@ static int ensure_engine(cam_id_t cam, char *err, size_t errlen)
} while (clients > 0 && ts_diff(&t, &t0) < SWITCH_GRACE_S); } while (clients > 0 && ts_diff(&t, &t0) < SWITCH_GRACE_S);
if (clients > 0) { if (clients > 0) {
snprintf(err, errlen, snprintf(err, errlen,
"camera busy: %d client(s) streaming %s", "camera switch timed out: %d client(s) still "
clients, camdefs[cur].name); "attached to %s", clients, camdefs[cur].name);
return -1; return -1;
} }
} }
@@ -878,6 +886,7 @@ int cam_snapshot(cam_id_t cam, int full, int quality,
struct cam_client { struct cam_client {
uint64_t last_seq; uint64_t last_seq;
uint64_t gen; /* kick generation at open; a bump ends the stream */
uint8_t *buf; uint8_t *buf;
size_t cap; size_t cap;
}; };
@@ -897,6 +906,7 @@ cam_client_t *cam_client_open(cam_id_t cam, char *err, size_t errlen)
} }
pthread_mutex_lock(&eng.lock); pthread_mutex_lock(&eng.lock);
eng.clients++; eng.clients++;
c->gen = eng.kick_gen;
now_ts(&eng.last_activity); now_ts(&eng.last_activity);
pthread_mutex_unlock(&eng.lock); pthread_mutex_unlock(&eng.lock);
pthread_mutex_unlock(&eng.ctl); pthread_mutex_unlock(&eng.ctl);
@@ -907,7 +917,7 @@ long cam_client_next(cam_client_t *c, const uint8_t **jpeg)
{ {
pthread_mutex_lock(&eng.lock); pthread_mutex_lock(&eng.lock);
int timeouts = 0; int timeouts = 0;
while (eng.running && eng.seq <= c->last_seq) { while (eng.running && c->gen == eng.kick_gen && eng.seq <= c->last_seq) {
struct timespec deadline; struct timespec deadline;
clock_gettime(CLOCK_REALTIME, &deadline); clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += CLIENT_WAIT_S; deadline.tv_sec += CLIENT_WAIT_S;
@@ -915,7 +925,7 @@ long cam_client_next(cam_client_t *c, const uint8_t **jpeg)
== ETIMEDOUT && ++timeouts >= 2) == ETIMEDOUT && ++timeouts >= 2)
break; break;
} }
if (!eng.running || eng.seq <= c->last_seq) { if (!eng.running || c->gen != eng.kick_gen || eng.seq <= c->last_seq) {
pthread_mutex_unlock(&eng.lock); pthread_mutex_unlock(&eng.lock);
return -1; return -1;
} }
@@ -36,9 +36,11 @@ void cam_engine_shutdown(void);
int cam_snapshot(cam_id_t cam, int full, int quality, int cam_snapshot(cam_id_t cam, int full, int quality,
uint8_t **jpeg, size_t *len, char *err, size_t errlen); uint8_t **jpeg, size_t *len, char *err, size_t errlen);
/* Stream client: open pins the engine to a camera (starting or switching it /* Stream client: open makes the engine serve `cam` (starting it, or
* if needed), next blocks for a frame newer than the last one returned and * preempting the current clients and switching - last request wins; the
* copies it into a client-owned buffer, close releases the pin. */ * preempted clients' next() returns -1 so their streams end cleanly).
* next blocks for a frame newer than the last one returned and copies it
* into a client-owned buffer, close releases the pin. */
typedef struct cam_client cam_client_t; typedef struct cam_client cam_client_t;
cam_client_t *cam_client_open(cam_id_t cam, char *err, size_t errlen); cam_client_t *cam_client_open(cam_id_t cam, char *err, size_t errlen);
@@ -12,11 +12,11 @@
* GET /cam/snapshot?cam=&res=full|half&q= single JPEG (default full res) * GET /cam/snapshot?cam=&res=full|half&q= single JPEG (default full res)
* GET /cam/status JSON engine status * GET /cam/status JSON engine status
* *
* The two cameras share the hardware mux, so streaming clients pin the * The two cameras share the hardware mux; the newest request wins it. A
* selection: a STREAM request for the other camera waits a short grace * STREAM request for the other camera preempts the current stream
* for the pin to drain, then returns 409. A SNAPSHOT of the other camera * clients (their streams end cleanly - viewers freeze on the last frame)
* never fails busy - the engine borrows the mux for one frame and the * and switches. A SNAPSHOT of the other camera does not switch: the
* stream freezes for a few seconds instead. * engine borrows the mux for one frame and the stream freezes briefly.
* Environment: FORGECTRL_PORT (8080), FORGECTRL_STREAM_Q (75), * Environment: FORGECTRL_PORT (8080), FORGECTRL_STREAM_Q (75),
* FORGECTRL_LAMP (132). * FORGECTRL_LAMP (132).
* *
@@ -246,11 +246,17 @@ static const char index_html[] =
"msg=document.getElementById('msg');" "msg=document.getElementById('msg');"
"function setCam(c){cam=c;retries=0;msg.textContent='';" "function setCam(c){cam=c;retries=0;msg.textContent='';"
"v.src='/cam/stream?cam='+c+'&t='+Date.now();}" "v.src='/cam/stream?cam='+c+'&t='+Date.now();}"
"v.onerror=function(){if(retries++<5){" "function reload(){v.src='/cam/stream?cam='+cam+'&t='+Date.now();}"
"msg.textContent='stream retrying...';" /* Retry only when the engine still serves our camera - if another
"setTimeout(function(){v.src='/cam/stream?cam='+cam+'&t='+Date.now();}," * viewer preempted it, retrying would steal it right back. */
"700);}else{msg.textContent=" "v.onerror=function(){fetch('/cam/status').then(function(r){"
"'stream unavailable (camera busy from another viewer?)';}};" "return r.json();}).then(function(s){"
"if(s.cam===cam&&retries++<5){msg.textContent='stream retrying...';"
"setTimeout(reload,700);}else if(s.cam!==cam){msg.textContent="
"'stream taken by another viewer ('+s.cam+') - press a button to resume';}"
"else{msg.textContent='stream error - press a stream button to retry';}"
"}).catch(function(){msg.textContent="
"'service unreachable';});};"
"function peek(){" "function peek(){"
"msg.textContent='head peek (stream pauses a few seconds)...';" "msg.textContent='head peek (stream pauses a few seconds)...';"
"p.style.display='inline';p.onload=function(){msg.textContent='';};" "p.style.display='inline';p.onload=function(){msg.textContent='';};"