async.c File Reference

Go to the source code of this file.

Data Structures
struct	async_packet
struct	async_tcp_t

Macros
#define	ASYNC_RECEIVE_POOL_CAPACITY   (256u * 1024u)
#define	ASYNC_RECEIVE_POOL_SIZE   64u

Typedefs
typedef struct async_packet	async_packet_t

Functions
void	acip_async_tcp_transport_detach (acip_transport_t *transport)
bool	acip_async_tcp_transport_is (const acip_transport_t *transport)
size_t	acip_async_tcp_transport_queued (acip_transport_t *transport)
acip_transport_t *	acip_async_tcp_transport_create (platform_event_stream_t *stream, crypto_context_t *crypto)
acip_transport_t *	acip_async_tcp_client_create (socket_t *socket, crypto_context_t *crypto)
acip_transport_t *	acip_async_tcp_accepted_create (platform_event_stream_t *stream)

Macro Definition Documentation

ASYNC_RECEIVE_POOL_CAPACITY

#define ASYNC_RECEIVE_POOL_CAPACITY   (256u * 1024u)

Definition at line 25 of file async.c.

Referenced by acip_async_tcp_client_create().

ASYNC_RECEIVE_POOL_SIZE

#define ASYNC_RECEIVE_POOL_SIZE   64u

Definition at line 26 of file async.c.

Referenced by acip_async_tcp_client_create().

Typedef Documentation

async_packet_t

typedef struct async_packet async_packet_t

Function Documentation

acip_async_tcp_accepted_create()

acip_transport_t * acip_async_tcp_accepted_create

(

platform_event_stream_t *

stream

)

Queue complete packets from an accepted stream on its listener's loop. The returned transport is owned by the accepting caller. Destroy it outside the loop thread while the listener is running, or after the listener stops.

Definition at line 576 of file async.c.

                                                                                  {
  acip_transport_t *transport = acip_async_tcp_transport_create(stream, NULL);
  if (!transport)
    return NULL;
  async_tcp_t *tcp = transport->impl_data;
  if (cond_init(&tcp->received, "accepted_receive") != ASCIICHAT_OK) {
    platform_event_stream_observe_queue(tcp->stream, NULL, NULL);
    tcp->stream = NULL;
    atomic_store_bool(&tcp->closed, true);
    acip_transport_destroy(transport);
    return NULL;
  }
  tcp->queued_receive = true;
  asciichat_error_t result = platform_event_stream_start(stream, client_read, client_closed, tcp);
  if (result != ASCIICHAT_OK) {
    // The stream retains its callbacks on startup failure; release their state after close completes.
    tcp->destroy_on_close = transport;
    platform_event_stream_close(stream, result);
    return NULL;
  }
  return transport;
}

References acip_async_tcp_transport_create(), acip_transport_destroy(), ASCIICHAT_OK, async_tcp_t::closed, async_tcp_t::destroy_on_close, acip_transport::impl_data, platform_event_stream_close(), platform_event_stream_observe_queue(), platform_event_stream_start(), async_tcp_t::queued_receive, async_tcp_t::received, and async_tcp_t::stream.

acip_async_tcp_client_create()

acip_transport_t * acip_async_tcp_client_create	(	socket_t *	socket,
		crypto_context_t *	crypto )

Native client adapter. Owns a loop thread; recv waits on a bounded packet queue. Takes socket ownership when *socket becomes INVALID_SOCKET_VALUE.

Definition at line 525 of file async.c.

                                                                                           {
  acip_transport_t *transport = acip_async_tcp_transport_create(NULL, crypto);
  if (!transport)
    return NULL;
  async_tcp_t *tcp = transport->impl_data;
  if (cond_init(&tcp->received, "async_tcp_receive") != ASCIICHAT_OK) {
    acip_transport_destroy(transport);
    return NULL;
  }
  tcp->queued_receive = true;
  tcp->receive_free = ringbuffer_create(sizeof(void *), ASYNC_RECEIVE_POOL_SIZE);
  if (!tcp->receive_free)
    goto client_pool_failure;
  for (size_t i = 0; i < ASYNC_RECEIVE_POOL_SIZE; i++) {
    void *pool_buffer = buffer_alloc(NULL, ASYNC_RECEIVE_POOL_CAPACITY);
    if (!pool_buffer)
      goto client_pool_failure;
    if (!ringbuffer_write(tcp->receive_free, &pool_buffer)) {
      buffer_release(pool_buffer);
      goto client_pool_failure;
    }
  }
  if (platform_event_loop_create(&tcp->loop) != ASCIICHAT_OK) {
    acip_transport_destroy(transport);
    return NULL;
  }
  asciichat_error_t result = platform_event_stream_adopt(tcp->loop, socket, &tcp->stream);
  if (result == ASCIICHAT_OK) {
    platform_event_stream_observe_queue(tcp->stream, queue_changed, tcp);
    result = platform_event_stream_start(tcp->stream, client_read, client_closed, tcp);
  }
  if (result == ASCIICHAT_OK)
    result = platform_event_loop_tick(tcp->loop, 50, client_tick, tcp);
  if (result == ASCIICHAT_OK)
    result = platform_thread_create(&tcp->thread, "tcp_io", client_loop, tcp);
  if (result != ASCIICHAT_OK) {
    acip_transport_destroy(transport);
    return NULL;
  }
  tcp->thread_started = true;
  if (acip_outbound_create(transport) != ASCIICHAT_OK) {
    acip_transport_destroy(transport);
    return NULL;
  }
  return transport;
 
client_pool_failure:
  acip_transport_destroy(transport);
  return NULL;
}

References acip_async_tcp_transport_create(), acip_outbound_create(), acip_transport_destroy(), ASCIICHAT_OK, ASYNC_RECEIVE_POOL_CAPACITY, ASYNC_RECEIVE_POOL_SIZE, buffer_alloc(), buffer_release, acip_transport::impl_data, async_tcp_t::loop, platform_event_loop_create(), platform_event_loop_tick(), platform_event_stream_adopt(), platform_event_stream_observe_queue(), platform_event_stream_start(), async_tcp_t::queued_receive, async_tcp_t::receive_free, async_tcp_t::received, ringbuffer_create(), ringbuffer_write(), async_tcp_t::stream, async_tcp_t::thread, and async_tcp_t::thread_started.

acip_async_tcp_transport_create()

acip_transport_t * acip_async_tcp_transport_create	(	platform_event_stream_t *	stream,
		crypto_context_t *	crypto )

Outbound ACIP adapter. Incoming packets are delivered by the connection's decoder. Destroy after the stream close callback, when workers have drained.

Definition at line 407 of file async.c.

                                                                                                             {
  acip_transport_t *transport = SAFE_CALLOC(1, sizeof(*transport), acip_transport_t *);
  async_tcp_t *tcp = SAFE_CALLOC(1, sizeof(*tcp), async_tcp_t *);
  if (mutex_init(&tcp->mutex, "tcp_packet_io_lifetime_guard") != ASCIICHAT_OK) {
    SAFE_FREE(tcp);
    SAFE_FREE(transport);
    return NULL;
  }
  if (mutex_init(&tcp->receive_mutex, "tcp_receive_queue") != ASCIICHAT_OK) {
    mutex_destroy(&tcp->mutex);
    SAFE_FREE(tcp);
    SAFE_FREE(transport);
    return NULL;
  }
  if (cond_init(&tcp->writable, "tcp_output_writable") != ASCIICHAT_OK) {
    mutex_destroy(&tcp->receive_mutex);
    mutex_destroy(&tcp->mutex);
    SAFE_FREE(tcp);
    SAFE_FREE(transport);
    return NULL;
  }
  tcp->receive_queue = ringbuffer_create(sizeof(async_packet_t *), 4096);
  if (!tcp->receive_queue) {
    ringbuffer_destroy(tcp->receive_queue);
    cond_destroy(&tcp->writable);
    mutex_destroy(&tcp->receive_mutex);
    mutex_destroy(&tcp->mutex);
    SAFE_FREE(tcp);
    SAFE_FREE(transport);
    return NULL;
  }
  tcp->stream = stream;
  transport->methods = &async_methods;
  transport->impl_data = tcp;
  transport->crypto_ctx = crypto;
  transport->stats_peer = stats_runtime_peer_open("TCP-async");
  if (acip_outbound_create(transport) != ASCIICHAT_OK) {
    acip_transport_destroy(transport);
    return NULL;
  }
  platform_event_stream_observe_queue(stream, queue_changed, tcp);
  return transport;
}

References acip_outbound_create(), acip_transport_destroy(), ASCIICHAT_OK, acip_transport::crypto_ctx, acip_transport::impl_data, acip_transport::methods, async_tcp_t::mutex, platform_event_stream_observe_queue(), async_tcp_t::receive_mutex, async_tcp_t::receive_queue, ringbuffer_create(), ringbuffer_destroy(), SAFE_CALLOC, SAFE_FREE, acip_transport::stats_peer, stats_runtime_peer_open(), async_tcp_t::stream, and async_tcp_t::writable.

Referenced by acip_async_tcp_accepted_create(), and acip_async_tcp_client_create().

acip_async_tcp_transport_detach()

void acip_async_tcp_transport_detach

(

acip_transport_t *

transport

)

Definition at line 262 of file async.c.

                                                                  {
  async_tcp_t *tcp = transport->impl_data;
  mutex_lock(&tcp->mutex);
  atomic_store_bool(&tcp->closed, true);
  platform_event_stream_observe_queue(tcp->stream, NULL, NULL);
  tcp->stream = NULL;
  mutex_unlock(&tcp->mutex);
}

References async_tcp_t::closed, acip_transport::impl_data, async_tcp_t::mutex, platform_event_stream_observe_queue(), and async_tcp_t::stream.

acip_async_tcp_transport_is()

bool acip_async_tcp_transport_is

(

const acip_transport_t *

transport

)

Definition at line 388 of file async.c.

                                                                    {
  return transport && transport->methods == &async_methods;
}

References acip_transport::methods.

acip_async_tcp_transport_queued()

size_t acip_async_tcp_transport_queued

(

acip_transport_t *

transport

)

Definition at line 392 of file async.c.

                                                                    {
  async_tcp_t *tcp = transport->impl_data;
  return (size_t)atomic_load_u64(&tcp->queued_wire_bytes);
}

References acip_transport::impl_data, and async_tcp_t::queued_wire_bytes.