pipeline.c File Reference

Three-thread render pipeline implementation. More...

Go to the source code of this file.

Data Structures
struct	pipeline_frame_t
struct	frame_slot_s
struct	frame_queue_s
struct	session_pipeline_s
	For session pipeline, this is the live record that ties capture_tid, encode_tid, input_tid, input_start_status together. The owning code creates it, updates it while work is active, and releases its handles and pointed-to storage before freeing the record. More...

Macros
#define	PIPELINE_FRAME_SLOTS   3

Typedefs
typedef struct frame_slot_s	frame_slot_t
typedef struct frame_queue_s	frame_queue_t

Functions
asciichat_error_t	session_pipeline_create (session_capture_ctx_t *capture, session_display_ctx_t *display, session_pipeline_t **out)
asciichat_error_t	session_pipeline_run_main (session_pipeline_t *pipeline, session_should_exit_fn should_exit, session_keyboard_handler_fn keyboard_handler, void *user_data)
asciichat_error_t	session_pipeline_destroy (session_pipeline_t *pipeline)

Variables
bool	g_snapshot_first_frame_rendered

Detailed Description

Three-thread render pipeline implementation.

Definition in file pipeline.c.

Macro Definition Documentation

PIPELINE_FRAME_SLOTS

#define PIPELINE_FRAME_SLOTS   3

Definition at line 72 of file pipeline.c.

Referenced by session_pipeline_create().

Typedef Documentation

frame_queue_t

typedef struct frame_queue_s frame_queue_t

frame_slot_t

typedef struct frame_slot_s frame_slot_t

Definition at line 42 of file pipeline.c.

Function Documentation

session_pipeline_create()

asciichat_error_t session_pipeline_create	(	session_capture_ctx_t *	capture,
		session_display_ctx_t *	display,
		session_pipeline_t **	out )

Create and start the capture thread and (if –render-file) encode thread.

Parameters

capture	Capture context (must have real media_source_t for synchronous mode)
display	Display context (may have render_file set for –render-file mode)
out	Pointer to receive allocated pipeline context

Returns

ASCIICHAT_OK on success, error code otherwise

Note

The pipeline does NOT own capture or display — caller must keep them alive until session_pipeline_destroy() returns.

Definition at line 541 of file pipeline.c.

                                                                    {
  if (!capture || !display || !out) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "NULL argument");
  }
 
  session_pipeline_t *p = SAFE_CALLOC(1, sizeof(*p), session_pipeline_t *);
  for (size_t i = 0; i < PIPELINE_FRAME_SLOTS; ++i)
    atomic_init(&p->frame_slots[i].busy, false);
  p->capture = capture;
  p->display = display;
  if (platform_wakeup_create(&p->display_ready) != ASCIICHAT_OK) {
    SAFE_FREE(p);
    return SET_ERRNO(ERROR_INIT, "failed to create display wakeup");
  }
  atomic_ptr_store(&p->display_pending, NULL);
  NAMED_REGISTER_ATOMIC_PTR(&p->display_pending, "display.pending_frame", p);
  p->encode_queue = frame_queue_create(256, "encode"); // Large buffer for slow encoding
 
  // Check if display has render_file configured (will be checked in encode thread)
  // We store a flag to know whether to enqueue frames for encoding
  double render_fps = session_display_get_render_fps(display);
  bool has_file = session_display_has_render_file(display);
  media_source_t *source = session_capture_get_media_source(capture);
  // Synthetic/visualization sessions can switch sources too. Their recording
  // clock must remain elapsed session time across gaps and source FPS changes.
  if (media_source_get_type(source) != MEDIA_SOURCE_STDIN)
    session_display_set_render_live_timing(display);
 
  // Playback FPS also has a default when no recording was requested. It must
  // not enable the encoder's lossless queue for a live-only session.
  p->has_render_file = has_file;
  log_debug("[PIPELINE_CREATE] render_fps=%.1f, has_file=%d, has_render_file=%d", render_fps, has_file ? 1 : 0,
            p->has_render_file ? 1 : 0);
 
  atomic_store_bool(&p->stop, false);
  atomic_store_bool(&p->capture_finished, false);
  NAMED_REGISTER_ATOMIC(&p->capture_finished, "pipeline_capture_finished", p);
  atomic_store_u64(&p->first_frame_ns, 0);
 
  // Start capture thread
  if (platform_thread_create(&p->capture_tid, "pipeline_capture", pipeline_capture_thread, p) != 0) {
    free_frame(atomic_ptr_exchange(&p->display_pending, NULL));
    platform_wakeup_destroy(p->display_ready);
    NAMED_UNREGISTER(&p->display_pending);
    frame_queue_destroy(p->encode_queue);
    NAMED_UNREGISTER(&p->capture_finished);
    SAFE_FREE(p);
    return SET_ERRNO(ERROR_INIT, "failed to start capture thread");
  }
 
  // Start encode thread only if render_file is active
  if (p->has_render_file) {
    if (platform_thread_create(&p->encode_tid, "pipeline_encode", pipeline_encode_thread, p) != 0) {
      atomic_store_bool(&p->stop, true);
      thread_join(&p->capture_tid, NULL);
      free_frame(atomic_ptr_exchange(&p->display_pending, NULL));
      frame_slots_destroy(p->frame_slots);
      platform_wakeup_destroy(p->display_ready);
      NAMED_UNREGISTER(&p->display_pending);
      frame_queue_destroy(p->encode_queue);
      NAMED_UNREGISTER(&p->capture_finished);
      SAFE_FREE(p);
      return SET_ERRNO(ERROR_INIT, "failed to start encode thread");
    }
  }
 
  *out = p;
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, frame_slot_s::busy, session_pipeline_s::capture, session_pipeline_s::capture_finished, session_pipeline_s::capture_tid, session_pipeline_s::display, session_pipeline_s::display_pending, session_pipeline_s::display_ready, session_pipeline_s::encode_queue, session_pipeline_s::encode_tid, ERROR_INIT, ERROR_INVALID_PARAM, session_pipeline_s::first_frame_ns, session_pipeline_s::frame_slots, session_pipeline_s::has_render_file, log_debug, media_source_get_type(), MEDIA_SOURCE_STDIN, NAMED_REGISTER_ATOMIC, NAMED_REGISTER_ATOMIC_PTR, NAMED_UNREGISTER, out, PIPELINE_FRAME_SLOTS, platform_wakeup_create(), platform_wakeup_destroy(), SAFE_CALLOC, SAFE_FREE, session_capture_get_media_source(), session_display_get_render_fps(), session_display_has_render_file(), session_display_set_render_live_timing(), SET_ERRNO, source, and session_pipeline_s::stop.

Referenced by session_render_loop().

session_pipeline_destroy()

asciichat_error_t session_pipeline_destroy

(

session_pipeline_t *

pipeline

)

Stop threads and free pipeline.

Waits for capture and encode threads to exit cleanly.

Parameters

pipeline

Pipeline context (safe to call with NULL)

Returns

ASCIICHAT_OK

Definition at line 749 of file pipeline.c.

                                                                         {
  if (!pipeline)
    return ASCIICHAT_OK;
 
  log_debug("[PIPELINE] Destroying pipeline, waiting for threads...");
 
  // Signal threads to stop (may already be stopped)
  atomic_store_bool(&pipeline->stop, true);
  ascii_otel_pipeline_progress(3, 0);
 
  // Wake up any threads blocked on condition variable waits
  // so they can check the stop flag and exit immediately
  // instead of waiting for timeouts (e.g., 100ms frame_queue_pop timeout)
  platform_wakeup_signal(pipeline->display_ready);
  if (pipeline->encode_queue) {
    mutex_lock(&pipeline->encode_queue->mu);
    cond_broadcast(&pipeline->encode_queue->not_empty);
    cond_broadcast(&pipeline->encode_queue->not_full);
    mutex_unlock(&pipeline->encode_queue->mu);
  }
 
  if (thread_is_initialized(&pipeline->input_tid) && thread_join(&pipeline->input_tid, NULL) != ASCIICHAT_OK)
    return SET_ERRNO(ERROR_THREAD, "Cannot join session input thread during shutdown");
  session_display_begin_drain(pipeline->display, 0, false);
 
  // A timeout must never let the owner free memory still referenced by a worker.
  // Capture observes stop; encode drains its queue and exits once capture is done.
  if (thread_is_initialized(&pipeline->capture_tid)) {
    if (thread_join(&pipeline->capture_tid, NULL) != 0)
      return SET_ERRNO(ERROR_THREAD, "Cannot join pipeline capture thread");
  }
  session_display_begin_drain(pipeline->display, pipeline->frames_accepted, true);
  if (thread_is_initialized(&pipeline->encode_tid)) {
    if (thread_join(&pipeline->encode_tid, NULL) != 0)
      return SET_ERRNO(ERROR_THREAD, "Cannot join pipeline encode thread");
  }
 
  // Flush queues to free any remaining frames
  free_frame(atomic_ptr_exchange(&pipeline->display_pending, NULL));
  frame_queue_flush(pipeline->encode_queue, free_frame_generic);
 
  frame_slots_destroy(pipeline->frame_slots);
  platform_wakeup_destroy(pipeline->display_ready);
  NAMED_UNREGISTER(&pipeline->display_pending);
  frame_queue_destroy(pipeline->encode_queue);
  NAMED_UNREGISTER(&pipeline->capture_finished);
  asciichat_error_t err = pipeline->capture_failure.code != ASCIICHAT_OK ? errno_import(&pipeline->capture_failure)
                                                                         : pipeline->encode_error;
  if (pipeline->capture_failure.code != ASCIICHAT_OK)
    LOG_ERRNO_IF_SET("Media capture failed");
  SAFE_FREE(pipeline);
 
  log_debug("[PIPELINE] Pipeline destroyed");
  return err;
}

References ascii_otel_pipeline_progress(), ASCIICHAT_OK, session_pipeline_s::capture_failure, session_pipeline_s::capture_finished, session_pipeline_s::capture_tid, asciichat_error_context_t::code, session_pipeline_s::display, session_pipeline_s::display_pending, session_pipeline_s::display_ready, session_pipeline_s::encode_error, session_pipeline_s::encode_queue, session_pipeline_s::encode_tid, errno_import(), ERROR_THREAD, session_pipeline_s::frame_slots, session_pipeline_s::frames_accepted, session_pipeline_s::input_tid, log_debug, LOG_ERRNO_IF_SET, frame_queue_s::mu, NAMED_UNREGISTER, frame_queue_s::not_empty, frame_queue_s::not_full, platform_wakeup_destroy(), platform_wakeup_signal(), SAFE_FREE, session_display_begin_drain(), SET_ERRNO, and session_pipeline_s::stop.

Referenced by session_render_loop().

session_pipeline_run_main()

asciichat_error_t session_pipeline_run_main	(	session_pipeline_t *	pipeline,
		session_should_exit_fn	should_exit,
		session_keyboard_handler_fn	keyboard_handler,
		void *	user_data )

Main thread loop: drives terminal output and keyboard handling.

Pops frames from display queue, converts to ASCII, writes to terminal. Handles snapshot delay timer and exit conditions.

Parameters

pipeline	Pipeline context created by session_pipeline_create()
should_exit	Callback to check if rendering should stop
keyboard_handler	Optional keyboard handler callback
user_data	Opaque data passed to callbacks

Returns

ASCIICHAT_OK on success

Note

This function blocks until should_exit() returns true or capture ends. Signals capture/encode threads to stop before returning.

Definition at line 630 of file pipeline.c.

                                                                                                           {
  if (!pipeline || !should_exit) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "NULL argument");
  }
 
  log_debug("[PIPELINE_MAIN] Starting main thread loop");
  uint64_t display_epoch = UINT64_MAX;
  if (keyboard_handler && !GET_OPTION(snapshot_mode) && !(pipeline->has_render_file && !terminal_is_interactive())) {
    pipeline->keyboard_handler = keyboard_handler;
    pipeline->should_exit = should_exit;
    pipeline->input_data = user_data;
    atomic_store(&pipeline->input_start_status, 0);
    if (platform_thread_create(&pipeline->input_tid, "pipeline_input", pipeline_input_thread, pipeline) != ASCIICHAT_OK)
      return SET_ERRNO(ERROR_THREAD, "Cannot start responsive session input");
    while (atomic_load(&pipeline->input_start_status) == 0)
      platform_sleep_ms(1);
    if (atomic_load(&pipeline->input_start_status) < 0) {
      thread_join(&pipeline->input_tid, NULL);
      return SET_ERRNO(ERROR_INIT, "Cannot initialize session keyboard input");
    }
  }
 
  // Snapshot duration is owned by the capture thread. It records the first
  // captured frame and marks capture_finished at its duration limit.
  // The display thread waits for producer completion instead of using
  // a second timer based on terminal rendering speed; large terminals can make
  // ASCII conversion and render-file encoding substantially slower than capture.
  while (!should_exit(user_data) && !atomic_load_bool(&pipeline->stop)) {
    bool exit_check = should_exit(user_data);
    log_debug_every(US_PER_SEC_INT, "[PIPELINE_DEBUG] should_exit=%d, stop=%d", exit_check,
                    atomic_load_bool(&pipeline->stop));
 
    platform_wakeup_wait(pipeline->display_ready, 100 * NS_PER_MS_INT);
    pipeline_frame_t *frame = atomic_ptr_exchange(&pipeline->display_pending, NULL);
 
    if (!frame) {
      if (atomic_load_bool(&pipeline->capture_finished) && !atomic_ptr_load(&pipeline->display_pending))
        break;
      continue;
    }
 
    ascii_otel_metric(ASCII_OTEL_METRIC_DISPLAY_DEPTH, 0);
    double queue_age = (double)(time_get_ns() - frame->captured_ns) / 1e9;
    ascii_otel_metric(ASCII_OTEL_METRIC_DISPLAY_WAIT, queue_age);
    ascii_otel_metric(ASCII_OTEL_METRIC_QUEUE_OLDEST, queue_age);
    ASCII_OTEL_SCOPE_PARENT(PIPELINE_DISPLAY, &frame->telemetry_context);
    ascii_otel_scope_int(&otel_scope, "ascii_chat.queue.wait_ns", (int64_t)(time_get_ns() - frame->captured_ns));
    // Validate frame structure exists before accessing fields
    if (!frame_is_valid(frame)) {
      ascii_otel_metric(ASCII_OTEL_METRIC_DROP_INVALID, 1);
      log_error("Received invalid frame: corrupted or NULL");
      free_frame(frame);
      frame = NULL;
      continue;
    }
 
    // Check if help screen is active - if so, don't render ASCII frames
    int screen = ui_controller_current_screen();
    // Snapshot mode has no interactive prompt/log overlay.  Do not discard its
    // only captured frame because startup UI state is still settling.
    bool help_is_active = !GET_OPTION(snapshot_mode) && (screen == UI_SCREEN_PROMPT || screen == UI_SCREEN_LOGS);
 
    if (!help_is_active) {
      // Convert to ASCII
      if (display_epoch != frame->source_epoch) {
        session_display_reset_stabilizer(pipeline->display, false);
        display_epoch = frame->source_epoch;
      }
      image_t tmp = {.w = frame->w, .h = frame->h, .pixels = (rgb_pixel_t *)frame->pixels};
      uint64_t conversion_started_ns = time_get_ns();
      char *ascii = session_display_convert_to_ascii_reuse(pipeline->display, &tmp);
      uint64_t converted_ns = time_get_ns();
      double read_ms = (double)(frame->captured_ns - frame->acquisition_started_ns) / NS_PER_MS_INT;
      double queue_ms = (double)(conversion_started_ns - frame->captured_ns) / NS_PER_MS_INT;
      double convert_ms = (double)(converted_ns - conversion_started_ns) / NS_PER_MS_INT;
      uint64_t epoch = frame->source_epoch;
      bool fresh = frame->fresh;
      uint64_t acquisition_started_ns = frame->acquisition_started_ns;
      free_frame(frame);
      frame = NULL;
 
      if (ascii) {
        if (epoch == session_capture_source_epoch(pipeline->capture)) {
          ui_controller_frame_acquired_at(acquisition_started_ns);
          session_display_write_ascii_frame(pipeline->display, ascii, fresh);
          ui_controller_frame_acquired_at(0);
          uint64_t published_ns = time_get_ns();
          ascii_otel_metric(ASCII_OTEL_METRIC_FRAME_DISPLAYED, 1);
          ascii_otel_pipeline_progress(1, 0);
          ascii_otel_metric(ASCII_OTEL_METRIC_FRAME_LATENCY, (double)(published_ns - acquisition_started_ns) / 1e9);
          session_capture_complete_open(pipeline->capture, epoch);
          log_debug_every(US_PER_SEC_INT,
                          "LIVE_FRAME: capture_to_publish_ms=%.2f fresh=%d read_ms=%.2f queue_ms=%.2f convert_ms=%.2f "
                          "publish_ms=%.2f",
                          (double)(published_ns - acquisition_started_ns) / NS_PER_MS_INT, fresh, read_ms, queue_ms,
                          convert_ms, (double)(published_ns - converted_ns) / NS_PER_MS_INT);
        }
        session_display_release_ascii(pipeline->display, ascii);
      } else {
        ascii_otel_metric(ASCII_OTEL_METRIC_DROP_CONVERT, 1);
        log_error_every(US_PER_SEC_INT, "Failed to convert captured frame to ASCII; skipping display frame");
      }
    } else {
      ascii_otel_metric(ASCII_OTEL_METRIC_DROP_SCREEN, 1);
      free_frame(frame);
      frame = NULL;
    }
  }
 
  log_debug("[PIPELINE_MAIN] Main loop exiting, signaling threads to stop");
  atomic_store_bool(&pipeline->stop, true);
  ascii_otel_pipeline_progress(3, 0);
  if (thread_is_initialized(&pipeline->input_tid) && thread_join(&pipeline->input_tid, NULL) != ASCIICHAT_OK)
    return SET_ERRNO(ERROR_THREAD, "Cannot join session input thread");
 
  return ASCIICHAT_OK;
}

References pipeline_frame_t::acquisition_started_ns, ascii_otel_metric(), ascii_otel_pipeline_progress(), ascii_otel_scope_int(), ASCII_OTEL_SCOPE_PARENT, ASCIICHAT_OK, session_pipeline_s::capture, session_pipeline_s::capture_finished, pipeline_frame_t::captured_ns, session_pipeline_s::display, session_pipeline_s::display_pending, session_pipeline_s::display_ready, ERROR_INIT, ERROR_INVALID_PARAM, ERROR_THREAD, pipeline_frame_t::fresh, GET_OPTION, pipeline_frame_t::h, session_pipeline_s::has_render_file, session_pipeline_s::input_data, session_pipeline_s::input_start_status, session_pipeline_s::input_tid, session_pipeline_s::keyboard_handler, log_debug, log_debug_every, log_error, log_error_every, NS_PER_MS_INT, pipeline_frame_t::pixels, platform_sleep_ms(), platform_wakeup_wait(), session_capture_complete_open(), session_capture_source_epoch(), session_display_convert_to_ascii_reuse(), session_display_release_ascii(), session_display_reset_stabilizer(), session_display_write_ascii_frame(), SET_ERRNO, session_pipeline_s::should_exit, should_exit(), pipeline_frame_t::source_epoch, session_pipeline_s::stop, pipeline_frame_t::telemetry_context, ui_controller_current_screen(), ui_controller_frame_acquired_at(), UI_SCREEN_LOGS, UI_SCREEN_PROMPT, US_PER_SEC_INT, and pipeline_frame_t::w.

Referenced by session_render_loop().

Variable Documentation

g_snapshot_first_frame_rendered

bool g_snapshot_first_frame_rendered

extern

Definition at line 54 of file src/common/session/render.c.

Referenced by session_display_write_ascii_frame(), and session_render_loop().