backpressure.c File Reference

Go to the source code of this file.

Data Structures
struct	outbound_packet
struct	outbound_stream_t
struct	acip_outbound

Macros
#define	OUTBOUND_MAX_PACKETS   512
#define	OUTBOUND_MAX_STREAMS   64
#define	WIRE_OVERHEAD   (sizeof(packet_header_t) + 32 + CRYPTO_NONCE_SIZE + CRYPTO_MAC_SIZE)

Typedefs
typedef struct outbound_packet	outbound_packet_t

Functions
asciichat_error_t	acip_outbound_create (acip_transport_t *transport)
void	acip_outbound_stop (acip_transport_t *transport)
void	acip_outbound_destroy (acip_transport_t *transport)
asciichat_error_t	acip_outbound_submitv (acip_transport_t *transport, const socket_buffer_t *buffers, size_t count, acip_admission_t *admission)
asciichat_error_t	acip_outbound_submitv_ex (acip_transport_t *transport, const socket_buffer_t *buffers, size_t count, const acip_send_options_t *options, acip_admission_t *admission)
asciichat_error_t	acip_outbound_snapshot (acip_transport_t *transport, acip_outbound_stats_t *stats)
bool	acip_outbound_needs_keyframe (acip_transport_t *transport, uint16_t type, uint32_t id)

Macro Definition Documentation

OUTBOUND_MAX_PACKETS

#define OUTBOUND_MAX_PACKETS   512

Definition at line 14 of file backpressure.c.

OUTBOUND_MAX_STREAMS

#define OUTBOUND_MAX_STREAMS   64

Definition at line 15 of file backpressure.c.

WIRE_OVERHEAD

#define WIRE_OVERHEAD   (sizeof(packet_header_t) + 32 + CRYPTO_NONCE_SIZE + CRYPTO_MAC_SIZE)

Definition at line 16 of file backpressure.c.

Typedef Documentation

outbound_packet_t

typedef struct outbound_packet outbound_packet_t

Function Documentation

acip_outbound_create()

asciichat_error_t acip_outbound_create

(

acip_transport_t *

transport

)

Definition at line 439 of file backpressure.c.

                                                                    {
  ASCII_OTEL_SCOPE(ACIP_OUTBOUND_CREATE);
  if (!transport || !transport->methods || (!transport->methods->send && !transport->methods->sendv_ex))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound backend");
  if (transport->outbound)
    return ASCIICHAT_OK;
  acip_outbound_t *out = SAFE_CALLOC(1, sizeof(*out), acip_outbound_t *);
  if (mutex_init(&out->mutex, "connection_outgoing_packet_queue") != ASCIICHAT_OK) {
    SAFE_FREE(out);
    return GET_ERRNO();
  }
  if (cond_init(&out->changed, "outgoing_packet_ready_or_done") != ASCIICHAT_OK) {
    mutex_destroy(&out->mutex);
    SAFE_FREE(out);
    return GET_ERRNO();
  }
  out->transport = transport;
  asciichat_error_t result = platform_thread_create(&out->worker, "outbound", outbound_worker, out);
  if (result != ASCIICHAT_OK) {
    cond_destroy(&out->changed);
    mutex_destroy(&out->mutex);
    SAFE_FREE(out);
    return result;
  }
  transport->outbound = out;
  return ASCIICHAT_OK;
}

References ASCII_OTEL_SCOPE, ASCIICHAT_OK, ERROR_INVALID_PARAM, GET_ERRNO, acip_transport::methods, out, acip_transport::outbound, SAFE_CALLOC, SAFE_FREE, acip_transport_methods_t::send, acip_transport_methods_t::sendv_ex, and SET_ERRNO.

Referenced by acip_async_tcp_client_create(), acip_async_tcp_transport_create(), acip_tcp_transport_create(), acip_websocket_client_transport_create(), acip_websocket_server_transport_create(), and quic_transport_begin_security().

acip_outbound_destroy()

void acip_outbound_destroy

(

acip_transport_t *

transport

)

Stop admission, interrupt writes, join the worker, and release every queued owner.

Definition at line 478 of file backpressure.c.

                                                        {
  ASCII_OTEL_SCOPE(ACIP_OUTBOUND_DESTROY);
  acip_outbound_t *out = transport->outbound;
  if (!out)
    return;
  mutex_lock(&out->mutex);
  out->stopping = true;
  cond_broadcast(&out->changed);
  mutex_unlock(&out->mutex);
  /* Wake a blocked send while every backend resource is still alive. The
   * failed-send worker also closes its backend, so wait for that worker before
   * closing here: two concurrent closes can join the same service thread. */
  if (transport->methods->interrupt)
    transport->methods->interrupt(transport);
  else
    transport->methods->close(transport);
  thread_join(&out->worker, NULL);
  if (transport->methods->interrupt)
    transport->methods->close(transport);
  mutex_lock(&out->mutex);
  outbound_record_failure_and_fail_pending(out, ERROR_NETWORK, NULL, NULL);
  while (out->recycled) {
    outbound_packet_t *p = out->recycled;
    out->recycled = p->next;
    packet_free(out, p);
  }
  mutex_unlock(&out->mutex);
  cond_destroy(&out->changed);
  mutex_destroy(&out->mutex);
  SAFE_FREE(transport->outbound);
}

References ASCII_OTEL_SCOPE, acip_transport_methods_t::close, ERROR_NETWORK, acip_transport_methods_t::interrupt, acip_transport::methods, outbound_packet::next, out, acip_transport::outbound, and SAFE_FREE.

Referenced by acip_transport_destroy().

acip_outbound_needs_keyframe()

bool acip_outbound_needs_keyframe	(	acip_transport_t *	transport,
		uint16_t	type,
		uint32_t	id )

Definition at line 709 of file backpressure.c.

                                                                                           {
  if (!transport || !transport->outbound)
    return false;
  acip_outbound_t *out = transport->outbound;
  if (transport->methods->consume_keyframe_request &&
      transport->methods->consume_keyframe_request(transport, type, id)) {
    outbound_packet_t stream = {.type = type, .stream = id, .video = true, .encoded = true};
    mutex_lock(&out->mutex);
    discard_stream(out, &stream);
    mutex_unlock(&out->mutex);
    return true;
  }
  mutex_lock(&out->mutex);
  size_t count = (size_t)atomic_load_u64(&out->stream_count);
  for (size_t i = 0; i < count; ++i)
    if (out->streams[i].type == type && out->streams[i].stream == id) {
      bool needs_keyframe = atomic_load_bool(&out->streams[i].needs_keyframe);
      mutex_unlock(&out->mutex);
      return needs_keyframe;
    }
  mutex_unlock(&out->mutex);
  return true;
}

References acip_transport_methods_t::consume_keyframe_request, acip_transport::methods, out, and acip_transport::outbound.

Referenced by acip_send_image_frame_h265(), and threaded_send_image_frame_encoded().

acip_outbound_snapshot()

asciichat_error_t acip_outbound_snapshot	(	acip_transport_t *	transport,
		acip_outbound_stats_t *	stats )

Definition at line 687 of file backpressure.c.

                                                                                                    {
  if (!transport || !transport->outbound || !stats)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound statistics");
  acip_outbound_t *out = transport->outbound;
  mutex_lock(&out->mutex);
  stats->accepted = atomic_load_u64(&out->accepted);
  stats->submitted = atomic_load_u64(&out->submitted);
  stats->dropped = atomic_load_u64(&out->dropped);
  stats->congestion = atomic_load_u64(&out->congestion);
  stats->recovery_frames = atomic_load_u64(&out->recovery_frames);
  stats->peak_bytes = (size_t)atomic_load_u64(&out->peak_bytes);
  stats->application_bytes = out->bytes + out->active_bytes;
  stats->backend_bytes = backend_bytes(out);
  stats->backend_reserved_bytes = transport->methods->reserved_bytes;
  uint64_t now = time_get_ns();
  for (outbound_packet_t *p = out->head; p; p = p->next)
    if (p->video && now - p->created > stats->oldest_frame_age_ns)
      stats->oldest_frame_age_ns = now - p->created;
  mutex_unlock(&out->mutex);
  return ASCIICHAT_OK;
}

References acip_outbound_stats_t::accepted, acip_outbound_stats_t::application_bytes, ASCIICHAT_OK, acip_outbound_stats_t::backend_bytes, acip_outbound_stats_t::backend_reserved_bytes, acip_outbound_stats_t::congestion, outbound_packet::created, acip_outbound_stats_t::dropped, ERROR_INVALID_PARAM, acip_transport::methods, outbound_packet::next, acip_outbound_stats_t::oldest_frame_age_ns, out, acip_transport::outbound, acip_outbound_stats_t::peak_bytes, acip_outbound_stats_t::recovery_frames, acip_transport_methods_t::reserved_bytes, SET_ERRNO, acip_outbound_stats_t::submitted, and outbound_packet::video.

acip_outbound_stop()

void acip_outbound_stop

(

acip_transport_t *

transport

)

Stop admission and wake queued producers without joining the worker or releasing backend resources.

Definition at line 467 of file backpressure.c.

                                                     {
  acip_outbound_t *out = transport ? transport->outbound : NULL;
  if (!out)
    return;
  mutex_lock(&out->mutex);
  out->stopping = true;
  out->stopped_by_caller = true;
  outbound_record_failure_and_fail_pending(out, ERROR_NETWORK, NULL, NULL);
  mutex_unlock(&out->mutex);
}

References ERROR_NETWORK, out, and acip_transport::outbound.

acip_outbound_submitv()

asciichat_error_t acip_outbound_submitv	(	acip_transport_t *	transport,
		const socket_buffer_t *	buffers,
		size_t	count,
		acip_admission_t *	admission )

Copy plaintext before returning ACCEPTED; congestion never changes encryption state. Handshake/control submissions except ping/pong wait for backend submission so crypto/setup barriers hold. Success describes local admission/submission, never receiver delivery. A terminal backend failure retains its status and errno context for waiting and subsequent producers. Congestion remains a nonfatal admission outcome.

Definition at line 675 of file backpressure.c.

                                                                     {
  return outbound_submitv(transport, buffers, count, NULL, admission);
}

acip_outbound_submitv_ex()

asciichat_error_t acip_outbound_submitv_ex	(	acip_transport_t *	transport,
		const socket_buffer_t *	buffers,
		size_t	count,
		const acip_send_options_t *	options,
		acip_admission_t *	admission )

Definition at line 680 of file backpressure.c.

                                                                                                            {
  if (!options)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound send options");
  return outbound_submitv(transport, buffers, count, options, admission);
}

References ERROR_INVALID_PARAM, and SET_ERRNO.

Referenced by acip_transport_sendv_ex().