From e4b6e8e7b617e98ee6195d1ec51f57a71f600cb6 Mon Sep 17 00:00:00 2001 From: simonCatBot Date: Tue, 28 Jul 2026 19:02:38 -0700 Subject: [PATCH] fix(pipeline): eliminate double execution and worker idle race Two related bugs introduced by the streaming extension were causing the pipeline conformance tests (especially ManualSchedule/4 loop_count=1000) to hang or timeout in CI: 1. vxExecuteGraph ran every graph twice. The pipeup warm-up loop set steady_done after the first iteration and only broke on the next loop check, so non-pipeup graphs executed twice. Add ownGraphHasPipeup() and only request the extra steady iteration when the graph actually contains a node with pipeup_output_depth > 1. 2. vxWaitGraph could return while the worker was mid-batch. The pipeline worker decremented in_flight to 0 between back-to-back frames, creating a window where vxWaitGraph saw 'idle' even though the worker was about to start the next frame. Add worker_is_processing and only clear it / signal idle when the worker truly has no more work. Local tests: - GraphPipeline fast: 37/37 pass - GraphPipeline.ManualSchedule/4: passes - GraphStreaming: 24/24 pass - Graph.*: 233/233 pass --- sample/framework/vx_graph.c | 39 +++++++++++++++++++++++----- sample/framework/vx_graph_pipeline.c | 39 +++++++++++++++++++++++++--- sample/include/vx_internal.h | 2 ++ 3 files changed, 70 insertions(+), 10 deletions(-) diff --git a/sample/framework/vx_graph.c b/sample/framework/vx_graph.c index f2e9e30..394f11b 100644 --- a/sample/framework/vx_graph.c +++ b/sample/framework/vx_graph.c @@ -585,6 +585,7 @@ VX_API_ENTRY vx_graph VX_API_CALL vxCreateGraph(vx_context context) graph->worker_running = vx_false_e; graph->worker_stop = vx_false_e; graph->worker = 0; + graph->worker_is_processing = vx_false_e; graph->in_flight = 0; for (vx_uint32 i = 0; i < VX_INT_MAX_PARAMS; i++) { @@ -2473,6 +2474,20 @@ static vx_bool ownAnyNodeInPipeup(vx_graph graph) return vx_false_e; } +static vx_bool ownGraphHasPipeup(vx_graph graph) +{ + vx_uint32 i; + for (i = 0; i < graph->numNodes; i++) + { + vx_node node = graph->nodes[i]; + if (node == NULL || node->kernel == NULL) + continue; + if (node->kernel->pipeup_output_depth > 1) + return vx_true_e; + } + return vx_false_e; +} + static vx_bool ownIsPredecessorInPipeup(vx_graph graph, vx_node node) { vx_uint32 p; @@ -2546,12 +2561,13 @@ static vx_status vxExecuteGraph(vx_graph graph, vx_uint32 depth) } #ifdef OPENVX_USE_STREAMING - /* For non-streaming graphs with pipeup-output-depth nodes, run internal - * warm-up iterations while any node is still in pipeup, then run exactly - * one steady iteration before returning to the caller. This mirrors the - * rustVX behaviour expected by the GraphStreaming conformance tests. - * Streaming graphs always run a single iteration per call. */ + /* For non-streaming graphs that contain pipeup-output-depth nodes, run + * internal warm-up iterations while any node is still in pipeup, then run + * exactly one steady iteration before returning. Graphs without pipeup + * nodes, and streaming graphs, run exactly one iteration per call. */ vx_bool steady_done = vx_false_e; + vx_bool needs_steady = (graph->streaming_thread_running == vx_false_e && + ownGraphHasPipeup(graph) == vx_true_e); while (status == VX_SUCCESS && action != VX_ACTION_ABANDON) { vx_bool any_pipeup = ownAnyNodeInPipeup(graph); @@ -2726,7 +2742,12 @@ static vx_status vxExecuteGraph(vx_graph graph, vx_uint32 depth) if (graph->streaming_thread_running == vx_true_e) break; if (!any_pipeup) - steady_done = vx_true_e; + { + if (needs_steady) + steady_done = vx_true_e; + else + break; + } } #else } @@ -2850,10 +2871,14 @@ VX_API_ENTRY vx_status VX_API_CALL vxWaitGraph(vx_graph graph) if (graph->pipeline_configured == vx_true_e && graph->schedule_mode != VX_GRAPH_SCHEDULE_MODE_NORMAL) { - while (graph->in_flight > 0) + ownSemWait(&graph->pipe_lock); + while (graph->worker_is_processing || graph->in_flight > 0) { + ownSemPost(&graph->pipe_lock); ownWaitEvent(&graph->idle_event, VX_INT_FOREVER); + ownSemWait(&graph->pipe_lock); } + ownSemPost(&graph->pipe_lock); return VX_SUCCESS; } #endif diff --git a/sample/framework/vx_graph_pipeline.c b/sample/framework/vx_graph_pipeline.c index ef5cfea..19db816 100644 --- a/sample/framework/vx_graph_pipeline.c +++ b/sample/framework/vx_graph_pipeline.c @@ -186,13 +186,36 @@ static vx_value_t vxPipelineWorker(void *arg) { graph->in_flight++; if (graph->in_flight == 1) + { + graph->worker_is_processing = vx_true_e; ownResetEvent(&graph->idle_event); + } status = vxPipelineSwapRefs(graph); } + else if (graph->worker_is_processing == vx_true_e) + { + /* Batch is complete and the worker is about to go idle. */ + if (graph->in_flight > 0) + graph->in_flight--; + graph->worker_is_processing = vx_false_e; + ownSetEvent(&graph->idle_event); + } ownSemPost(&graph->pipe_lock); - if (can_run == vx_false_e || status != VX_SUCCESS) + if (can_run == vx_false_e) + break; + + if (status != VX_SUCCESS) + { + /* Swap failed after we already claimed an in-flight slot. */ + ownSemWait(&graph->pipe_lock); + if (graph->in_flight > 0) + graph->in_flight--; + graph->worker_is_processing = vx_false_e; + ownSetEvent(&graph->idle_event); + ownSemPost(&graph->pipe_lock); break; + } status = vxProcessGraph(graph); @@ -200,8 +223,6 @@ static vx_value_t vxPipelineWorker(void *arg) vxPipelineEnqueueDone(graph); vxPipelineRestoreRefs(graph); graph->in_flight--; - if (graph->in_flight == 0) - ownSetEvent(&graph->idle_event); ownSemPost(&graph->pipe_lock); if (graph->base.context->events_enabled == vx_true_e) @@ -246,6 +267,18 @@ static vx_value_t vxPipelineWorker(void *arg) ownPipelinePostEvent(graph->base.context, &event); } } + + /* If the worker was asked to stop while it still had an active batch, + * clean up the in-flight state so vxWaitGraph does not hang forever. */ + if (graph->worker_is_processing == vx_true_e) + { + ownSemWait(&graph->pipe_lock); + if (graph->in_flight > 0) + graph->in_flight--; + graph->worker_is_processing = vx_false_e; + ownSetEvent(&graph->idle_event); + ownSemPost(&graph->pipe_lock); + } } return 0; } diff --git a/sample/include/vx_internal.h b/sample/include/vx_internal.h index 5fbf5af..3cccaf8 100644 --- a/sample/include/vx_internal.h +++ b/sample/include/vx_internal.h @@ -1258,6 +1258,8 @@ typedef struct _vx_graph { vx_bool worker_stop; /*! \brief Counter of outstanding pipeline executions */ vx_int32 in_flight; + /*! \brief True while the pipeline worker is actively processing a batch */ + vx_bool worker_is_processing; /*! \brief Event signaled when worker becomes idle */ vx_internal_event_t idle_event; #endif