pipeline.h File Reference

Three-thread render pipeline: capture → display/encode. More...

Go to the source code of this file.

Typedefs
typedef struct session_capture_ctx	session_capture_ctx_t
typedef struct session_display_ctx	session_display_ctx_t
typedef struct session_pipeline_s	session_pipeline_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)

Detailed Description

Three-thread render pipeline: capture → display/encode.

Decouples capture, terminal display, and file encoding into separate threads:

• Capture thread: reads raw image_t frames from media source
• Main thread: converts to ASCII, prints to terminal (fast)
• Encode thread: converts to ASCII, encodes to FFmpeg file (slow, non-blocking)

Definition in file pipeline.h.

Typedef Documentation

session_capture_ctx_t

typedef struct session_capture_ctx session_capture_ctx_t

Definition at line 20 of file pipeline.h.

session_display_ctx_t

typedef struct session_display_ctx session_display_ctx_t

Definition at line 21 of file pipeline.h.

session_pipeline_t

typedef struct session_pipeline_s session_pipeline_t

Definition at line 22 of file pipeline.h.

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 420 of file pipeline.c.

                                                                    {
  if (!capture || !display || !out) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "session_pipeline_create: NULL argument");
  }
 
  session_pipeline_t *p = SAFE_CALLOC(1, sizeof(*p), session_pipeline_t *);
  p->capture = capture;
  p->display = display;
  p->display_queue = frame_queue_create(4, "display");
  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);
  if (media_source_uses_webcam(source) || media_source_get_type(source) == MEDIA_SOURCE_FILE)
    session_display_set_render_live_timing(display);
 
  // Enable encode thread if render_file is available
  // Note: has_file can be true even if render_file creation failed (ctx->render_file exists)
  // OR if render_fps is explicitly set for playback speed control
  p->has_render_file = render_fps > 0 || has_file;
  log_info("[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_u64(&p->first_frame_ns, 0);
 
  // Start capture thread
  if (asciichat_thread_create(&p->capture_tid, "pipeline_capture", pipeline_capture_thread, p) != 0) {
    frame_queue_destroy(p->display_queue);
    frame_queue_destroy(p->encode_queue);
    SAFE_FREE(p);
    return SET_ERRNO(ERROR_INIT, "session_pipeline_create: failed to start capture thread");
  }
 
  // Start encode thread only if render_file is active
  if (p->has_render_file) {
    if (asciichat_thread_create(&p->encode_tid, "pipeline_encode", pipeline_encode_thread, p) != 0) {
      atomic_store_bool(&p->stop, true);
      asciichat_thread_join(&p->capture_tid, NULL);
      frame_queue_destroy(p->display_queue);
      frame_queue_destroy(p->encode_queue);
      SAFE_FREE(p);
      return SET_ERRNO(ERROR_INIT, "session_pipeline_create: failed to start encode thread");
    }
  }
 
  *out = p;
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, asciichat_thread_create, asciichat_thread_join, atomic_store_bool(), atomic_store_u64(), session_pipeline_s::capture, session_pipeline_s::capture_tid, session_pipeline_s::display, session_pipeline_s::display_queue, session_pipeline_s::encode_queue, session_pipeline_s::encode_tid, ERROR_INIT, ERROR_INVALID_PARAM, session_pipeline_s::first_frame_ns, session_pipeline_s::has_render_file, log_info, MEDIA_SOURCE_FILE, media_source_get_type(), media_source_uses_webcam(), 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, 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 557 of file pipeline.c.

                                                                         {
  if (!pipeline)
    return ASCIICHAT_OK;
 
  log_info("[PIPELINE] Destroying pipeline, waiting for threads...");
 
  // Signal threads to stop (may already be stopped)
  atomic_store_bool(&pipeline->stop, true);
 
  // 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)
  if (pipeline->display_queue) {
    mutex_lock(&pipeline->display_queue->mu);
    cond_broadcast(&pipeline->display_queue->not_empty);
    cond_broadcast(&pipeline->display_queue->not_full);
    mutex_unlock(&pipeline->display_queue->mu);
  }
  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);
  }
 
  // Drain and wait for capture thread with 1 second timeout
  if (asciichat_thread_is_initialized(&pipeline->capture_tid)) {
    int join_result = asciichat_thread_join_timeout(&pipeline->capture_tid, NULL, 1000 * NS_PER_MS_INT);
    if (join_result != 0) {
      log_warn("[PIPELINE] Capture thread join timed out or failed (result=%d)", join_result);
    }
  }
 
  // Drain and wait for encode thread with 1 second timeout
  if (asciichat_thread_is_initialized(&pipeline->encode_tid)) {
    int join_result = asciichat_thread_join_timeout(&pipeline->encode_tid, NULL, 1000 * NS_PER_MS_INT);
    if (join_result != 0) {
      log_warn("[PIPELINE] Encode thread join timed out or failed (result=%d)", join_result);
    }
  }
 
  // Flush queues to free any remaining frames
  frame_queue_flush(pipeline->display_queue, free_frame_generic);
  frame_queue_flush(pipeline->encode_queue, free_frame_generic);
 
  // Don't free queue structures - debug_sync monitoring thread may still be accessing
  // the condition variables. Queues will be cleaned up with the pipeline structure.
  SAFE_FREE(pipeline);
 
  log_info("[PIPELINE] Pipeline destroyed");
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, asciichat_thread_is_initialized(), asciichat_thread_join_timeout(), atomic_store_bool(), session_pipeline_s::capture_tid, cond_broadcast(), session_pipeline_s::display_queue, session_pipeline_s::encode_queue, session_pipeline_s::encode_tid, log_info, log_warn, frame_queue_s::mu, mutex_lock, mutex_unlock, frame_queue_s::not_empty, frame_queue_s::not_full, NS_PER_MS_INT, SAFE_FREE, 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 474 of file pipeline.c.

                                                                                                           {
  if (!pipeline || !should_exit) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "session_pipeline_run_main: NULL argument");
  }
 
  log_info("[PIPELINE_MAIN] Starting main thread loop");
 
  // Snapshot duration is owned by the capture thread. It records the first
  // captured frame, runs until snapshot_delay has elapsed, then sends the EOF
  // sentinel. The display thread must wait for that sentinel 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(NS_PER_SEC_INT, "[PIPELINE_DEBUG] should_exit=%d, stop=%d", exit_check,
                    atomic_load_bool(&pipeline->stop));
 
    uint64_t pop_time_ns = time_get_ns();
    pipeline_frame_t *frame = (pipeline_frame_t *)frame_queue_pop(pipeline->display_queue, 1 * NS_PER_MS_INT);
 
    // Pause stops frame production, so keep polling the keyboard on queue
    // timeouts or there would be no way to resume playback.
    if (keyboard_handler) {
      keyboard_key_t key = keyboard_read_nonblocking();
      if (key != KEY_NONE) {
        log_debug("PIPELINE_KEYBOARD: Received key=%d", key);
        keyboard_handler(pipeline->capture, (int)key, user_data);
      }
    }
 
    if (!frame)
      continue; // timeout, check should_exit again
 
    // EOF sentinels are intentionally zero-initialized and therefore do not
    // satisfy normal frame validation.
    if (!frame->pixels) {
      log_info("[PIPELINE_MAIN_EOF] Received EOF sentinel, stopping");
      free_frame(frame);
      frame = NULL;
      break;
    }
 
    // Validate frame structure exists before accessing fields
    if (!frame_is_valid(frame)) {
      log_error("Received invalid frame: corrupted or NULL");
      free_frame(frame);
      frame = NULL;
      continue;
    }
 
    // Log frame pop with timestamp
    log_info("[PIPELINE_MAIN_POP] frame=%p, pixels=%p, w=%d, h=%d at %llu ns", (void *)frame, (void *)frame->pixels,
             frame->w, frame->h, (unsigned long long)pop_time_ns);
 
    // Check if help screen is active - if so, don't render ASCII frames
    bool help_is_active = pipeline->display && keyboard_help_is_active(pipeline->display);
 
    if (!help_is_active) {
      log_info("[PIPELINE_MAIN_RENDER] Converting frame to ASCII: %dx%d", frame->w, frame->h);
      // Convert to ASCII
      image_t tmp = {.w = frame->w, .h = frame->h, .pixels = (rgb_pixel_t *)frame->pixels};
      char *ascii = session_display_convert_to_ascii(pipeline->display, &tmp);
      free_frame(frame);
      frame = NULL;
 
      if (ascii) {
        // Write ASCII to terminal
        session_display_write_ascii(pipeline->display, ascii);
        SAFE_FREE(ascii);
      }
    } else {
      free_frame(frame);
      frame = NULL;
    }
  }
 
  log_info("[PIPELINE_MAIN] Main loop exiting, signaling threads to stop");
  atomic_store_bool(&pipeline->stop, true);
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, atomic_load_bool(), atomic_store_bool(), session_pipeline_s::capture, session_pipeline_s::display, session_pipeline_s::display_queue, ERROR_INVALID_PARAM, frame, KEY_NONE, keyboard_help_is_active(), keyboard_read_nonblocking(), log_debug, log_debug_every, log_error, log_info, NS_PER_MS_INT, NS_PER_SEC_INT, SAFE_FREE, session_display_convert_to_ascii(), session_display_write_ascii(), SET_ERRNO, should_exit(), session_pipeline_s::stop, time_get_ns(), and image_t::w.

Referenced by session_render_loop().