event.c File Reference

Go to the source code of this file.

Data Structures
struct	event_command_t
struct	platform_event_loop
struct	platform_event_stream
	The useful part of event write is in request, stream, done, arg: together they describe what is pending, who owns it, and how much data is valid. Keep the object with its parent operation and use the parent cleanup path. More...
struct	platform_event_listener
	platform event stream is intentionally a small, concrete record. Fields such as listener, owner, tcp, timer give callers the identity and current value they need. It is normally borrowed from, or owned by, the surrounding subsystem—not independently managed. More...
struct	event_write
struct	event_work_t
	event command keeps the values that move through the subsystem: fn, arg. Some members are pointers into caller- or subsystem-owned memory, so lifetime follows the API that populated them rather than the struct copy itself. More...
struct	loop_work_t
	This is the handoff point for event work. Callers fill work, stream, run, done before passing it on; consumers read the values during the operation and return ownership only where the member contract explicitly says they do. More...

Macros
#define	EVENT_COMMAND_LIMIT   1024
#define	EVENT_WRITE_LIMIT   (16U * 1024U * 1024U)
#define	EVENT_WRITE_COUNT_LIMIT   4096
#define	EVENT_READ_SIZE   65536

Typedefs
typedef struct event_write	event_write_t

Functions
void	platform_event_stream_close (platform_event_stream_t *stream, asciichat_error_t reason)
void	platform_event_listener_close (platform_event_listener_t *listener)
asciichat_error_t	platform_event_loop_create (platform_event_loop_t **out)
asciichat_error_t	platform_event_loop_run (platform_event_loop_t *loop)
asciichat_error_t	platform_event_loop_post (platform_event_loop_t *loop, platform_event_task_fn fn, void *arg)
void	platform_event_loop_stop (platform_event_loop_t *loop)
asciichat_error_t	platform_event_loop_destroy (platform_event_loop_t *loop)
const char *	platform_event_backend (void)
asciichat_error_t	platform_event_loop_tick (platform_event_loop_t *loop, uint64_t interval, platform_event_task_fn fn, void *arg)
asciichat_error_t	platform_event_stream_start (platform_event_stream_t *stream, platform_event_read_fn read, platform_event_close_fn close, void *arg)
void	platform_event_stream_pause (platform_event_stream_t *stream)
asciichat_error_t	platform_event_stream_resume (platform_event_stream_t *stream)
asciichat_error_t	platform_event_stream_deadline (platform_event_stream_t *stream, uint64_t timeout_ms)
asciichat_error_t	platform_event_stream_adopt (platform_event_loop_t *loop, socket_t *socket, platform_event_stream_t **out)
asciichat_error_t	platform_event_listen (platform_event_loop_t *loop, socket_t *socket, platform_event_accept_fn accept, void *arg, platform_event_listener_t **out)
void	platform_event_stream_set_zerocopy (platform_event_stream_t *stream, bool enabled)
void	platform_event_stream_request_close (platform_event_stream_t *stream, asciichat_error_t reason)
asciichat_error_t	platform_event_stream_write (platform_event_stream_t *stream, const socket_buffer_t *buffers, size_t count, platform_event_write_fn done, void *arg)
size_t	platform_event_stream_queued_bytes (const platform_event_stream_t *stream)
void	platform_event_stream_observe_queue (platform_event_stream_t *stream, platform_event_queue_fn observer, void *context)
asciichat_error_t	platform_event_loop_work (platform_event_loop_t *loop, platform_event_task_fn run, platform_event_task_fn done, void *arg)
asciichat_error_t	platform_event_stream_work (platform_event_stream_t *stream, platform_event_work_fn run, platform_event_work_done_fn done, void *arg)
asciichat_error_t	platform_event_stream_peer (platform_event_stream_t *stream, struct sockaddr_storage *address, socklen_t *length)

Macro Definition Documentation

EVENT_COMMAND_LIMIT

#define EVENT_COMMAND_LIMIT   1024

Definition at line 7 of file event.c.

Referenced by platform_event_loop_post().

EVENT_READ_SIZE

#define EVENT_READ_SIZE   65536

Definition at line 10 of file event.c.

EVENT_WRITE_COUNT_LIMIT

#define EVENT_WRITE_COUNT_LIMIT   4096

Definition at line 9 of file event.c.

Referenced by platform_event_stream_write().

EVENT_WRITE_LIMIT

#define EVENT_WRITE_LIMIT   (16U * 1024U * 1024U)

Definition at line 8 of file event.c.

Referenced by platform_event_stream_write().

Typedef Documentation

event_write_t

typedef struct event_write event_write_t

Function Documentation

platform_event_backend()

const char * platform_event_backend

(

void

)

Definition at line 279 of file event.c.

                                         {
#ifdef _WIN32
  return "iocp";
#elif defined(__linux__)
  return "epoll";
#elif defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
  return "kqueue";
#else
  return "libuv";
#endif
}

platform_event_listen()

asciichat_error_t platform_event_listen	(	platform_event_loop_t *	loop,
		socket_t *	socket,
		platform_event_accept_fn	accept,
		void *	arg,
		platform_event_listener_t **	out )

Sets *socket to INVALID_SOCKET_VALUE when ownership transfers, including if listen subsequently fails. Accepted streams are owned by the loop until their close callback returns. Set callbacks in the accept callback.

Definition at line 451 of file event.c.

                                                                                    {
  if (!loop || !socket || !accept || !out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid event listener");
  *out = NULL;
  platform_event_listener_t *listener = SAFE_CALLOC(1, sizeof(*listener), platform_event_listener_t *);
  int status = uv_tcp_init(&loop->uv, &listener->tcp);
  if (status < 0) {
    SAFE_FREE(listener);
    return event_error(status, "Initialize listener");
  }
  listener->tcp.data = listener;
  listener->listener = true;
  listener->accept = accept;
  listener->arg = arg;
  status = uv_tcp_open(&listener->tcp, (uv_os_sock_t)*socket);
  if (status < 0) {
    uv_close((uv_handle_t *)&listener->tcp, listener_closed);
    return event_error(status, "Adopt listener");
  }
  *socket = INVALID_SOCKET_VALUE;
  status = uv_listen((uv_stream_t *)&listener->tcp, 128, accepted);
  if (status < 0) {
    // Adoption has succeeded: the loop owns the descriptor even if listen fails.
    uv_close((uv_handle_t *)&listener->tcp, listener_closed);
    return event_error(status, "Listen");
  }
  *out = listener;
  return ASCIICHAT_OK;
}

References platform_event_listener::accept, platform_event_listener::arg, ASCIICHAT_OK, ERROR_INVALID_PARAM, platform_event_listener::listener, out, SAFE_CALLOC, SAFE_FREE, SET_ERRNO, platform_event_listener::tcp, and platform_event_loop::uv.

platform_event_listener_close()

void platform_event_listener_close

(

platform_event_listener_t *

listener

)

Definition at line 165 of file event.c.

                                                                        {
  if (listener && !uv_is_closing((uv_handle_t *)&listener->tcp))
    uv_close((uv_handle_t *)&listener->tcp, listener_closed);
}

References platform_event_listener::tcp.

platform_event_loop_create()

asciichat_error_t platform_event_loop_create

(

platform_event_loop_t **

out

)

Definition at line 205 of file event.c.

                                                                          {
  if (!out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing event loop output");
  *out = NULL;
  platform_event_loop_t *loop = SAFE_CALLOC(1, sizeof(*loop), platform_event_loop_t *);
  int status = uv_loop_init(&loop->uv);
  if (status < 0) {
    SAFE_FREE(loop);
    return event_error(status, "Initialize event loop");
  }
  status = uv_mutex_init(&loop->mutex);
  if (status < 0) {
    uv_loop_close(&loop->uv);
    SAFE_FREE(loop);
    return event_error(status, "Initialize event queue");
  }
  status = uv_async_init(&loop->uv, &loop->wake, wake_loop);
  if (status < 0) {
    uv_mutex_destroy(&loop->mutex);
    uv_loop_close(&loop->uv);
    SAFE_FREE(loop);
    return event_error(status, "Initialize event wakeup");
  }
  loop->wake.data = loop;
  loop->uv.data = loop;
  *out = loop;
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, platform_event_loop::mutex, out, SAFE_CALLOC, SAFE_FREE, SET_ERRNO, platform_event_loop::uv, and platform_event_loop::wake.

Referenced by acip_async_tcp_client_create().

platform_event_loop_destroy()

asciichat_error_t platform_event_loop_destroy

(

platform_event_loop_t *

loop

)

Definition at line 268 of file event.c.

                                                                           {
  if (!loop)
    return ASCIICHAT_OK;
  int status = uv_loop_close(&loop->uv);
  if (status < 0)
    return event_error(status, "Destroy event loop (stop and join first)");
  uv_mutex_destroy(&loop->mutex);
  SAFE_FREE(loop);
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, platform_event_loop::mutex, SAFE_FREE, and platform_event_loop::uv.

platform_event_loop_post()

asciichat_error_t platform_event_loop_post	(	platform_event_loop_t *	loop,
		platform_event_task_fn	fn,
		void *	arg )

Definition at line 241 of file event.c.

                                                                                                              {
  if (!loop || !fn)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing event command");
  uv_mutex_lock(&loop->mutex);
  if (loop->stopping || loop->count == EVENT_COMMAND_LIMIT) {
    uv_mutex_unlock(&loop->mutex);
    return SET_ERRNO(ERROR_INVALID_STATE, "Event queue stopped or full");
  }
  loop->commands[(loop->head + loop->count) % EVENT_COMMAND_LIMIT] = (event_command_t){fn, arg};
  ++loop->count;
  // Keep the wake handle alive until this send has finished.
  uv_async_send(&loop->wake);
  uv_mutex_unlock(&loop->mutex);
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, platform_event_loop::commands, platform_event_loop::count, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, EVENT_COMMAND_LIMIT, platform_event_loop::head, platform_event_loop::mutex, SET_ERRNO, platform_event_loop::stopping, and platform_event_loop::wake.

platform_event_loop_run()

asciichat_error_t platform_event_loop_run

(

platform_event_loop_t *

loop

)

Definition at line 234 of file event.c.

                                                                       {
  if (!loop)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing event loop");
  uv_run(&loop->uv, UV_RUN_DEFAULT);
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, SET_ERRNO, and platform_event_loop::uv.

platform_event_loop_stop()

void platform_event_loop_stop

(

platform_event_loop_t *

loop

)

Definition at line 257 of file event.c.

                                                           {
  if (!loop)
    return;
  uv_mutex_lock(&loop->mutex);
  if (!loop->stopping) {
    loop->stopping = true;
    uv_async_send(&loop->wake);
  }
  uv_mutex_unlock(&loop->mutex);
}

References platform_event_loop::mutex, platform_event_loop::stopping, and platform_event_loop::wake.

platform_event_loop_tick()

asciichat_error_t platform_event_loop_tick	(	platform_event_loop_t *	loop,
		uint64_t	interval,
		platform_event_task_fn	fn,
		void *	arg )

Definition at line 296 of file event.c.

                                                      {
  if (!loop || !interval || !fn || loop->tick_initialized)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid event loop tick");
  int status = uv_timer_init(&loop->uv, &loop->tick);
  if (status < 0)
    return event_error(status, "Initialize loop timer");
  loop->tick_initialized = true;
  loop->tick.data = loop;
  loop->tick_fn = fn;
  loop->tick_arg = arg;
  status = uv_timer_start(&loop->tick, tick_loop, interval, interval);
  return status < 0 ? event_error(status, "Start loop timer") : ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, SET_ERRNO, platform_event_loop::tick, platform_event_loop::tick_arg, platform_event_loop::tick_fn, platform_event_loop::tick_initialized, and platform_event_loop::uv.

Referenced by acip_async_tcp_client_create().

platform_event_loop_work()

asciichat_error_t platform_event_loop_work	(	platform_event_loop_t *	loop,
		platform_event_task_fn	run,
		platform_event_task_fn	done,
		void *	arg )

Definition at line 768 of file event.c.

                                                                                   {
  if (!loop || !run || !done)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid event worker");
  loop_work_t *work = SAFE_CALLOC(1, sizeof(*work), loop_work_t *);
  work->request.data = work;
  work->work = run;
  work->done = done;
  work->arg = arg;
  int status = uv_queue_work(&loop->uv, &work->request, loop_work_run, loop_work_done);
  if (status < 0) {
    SAFE_FREE(work);
    return event_error(status, "Queue loop work");
  }
  return ASCIICHAT_OK;
}

References loop_work_t::arg, ASCIICHAT_OK, loop_work_t::done, ERROR_INVALID_PARAM, loop_work_t::request, SAFE_CALLOC, SAFE_FREE, SET_ERRNO, platform_event_loop::uv, and loop_work_t::work.

platform_event_stream_adopt()

asciichat_error_t platform_event_stream_adopt	(	platform_event_loop_t *	loop,
		socket_t *	socket,
		platform_event_stream_t **	out )

Adopt an already-connected socket, setting it invalid on ownership transfer.

Definition at line 410 of file event.c.

                                                                             {
  if (!loop || !socket || !out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid socket adoption");
  *out = NULL;
  platform_event_stream_t *stream = create_stream(&loop->uv);
  if (!stream)
    return SET_ERRNO(ERROR_MEMORY, "Cannot initialize event stream");
  int status = uv_tcp_open(&stream->tcp, (uv_os_sock_t)*socket);
  if (status < 0) {
    platform_event_stream_close(stream, ERROR_NETWORK);
    return event_error(status, "Adopt TCP stream");
  }
  *socket = INVALID_SOCKET_VALUE;
  uv_tcp_nodelay(&stream->tcp, 1);
  uv_tcp_keepalive(&stream->tcp, 1, 60);
  *out = stream;
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_MEMORY, ERROR_NETWORK, out, platform_event_stream_close(), SET_ERRNO, event_write::stream, platform_event_stream::tcp, and platform_event_loop::uv.

Referenced by acip_async_tcp_client_create().

platform_event_stream_close()

void platform_event_stream_close	(	platform_event_stream_t *	stream,
		asciichat_error_t	reason )

Definition at line 120 of file event.c.

                                                                                            {
  if (!stream)
    return;
  uv_mutex_lock(&stream->mutex);
  if (stream->closing) {
    uv_mutex_unlock(&stream->mutex);
    return;
  }
  stream->closing = true;
  event_write_t *pending = stream->first;
  stream->first = stream->last = NULL;
  uv_mutex_unlock(&stream->mutex);
  while (pending) {
    event_write_t *next = pending->next;
    uv_mutex_lock(&stream->mutex);
    stream->queued -= pending->size;
    if (stream->queue_changed)
      stream->queue_changed(stream->queue_context, stream->queued);
    --stream->writes;
    uv_mutex_unlock(&stream->mutex);
    if (pending->done)
      pending->done(pending->arg, ERROR_NETWORK);
    socket_send_buffer_free(&pending->storage);
    SAFE_FREE(pending);
    pending = next;
  }
  stream->reason = reason;
  // Submitted uv_write requests complete during TCP close. A raw zerocopy
  // submission has no libuv request; discard its mapping without recycling it.
  if (stream->active && stream->active->waiting)
    write_complete(&stream->active->request, UV_ECANCELED);
  uv_timer_stop(&stream->output_timer);
  uv_close((uv_handle_t *)&stream->output_timer, stream_closed);
  uv_read_stop((uv_stream_t *)&stream->tcp);
  uv_timer_stop(&stream->timer);
  uv_close((uv_handle_t *)&stream->timer, stream_closed);
  uv_close((uv_handle_t *)&stream->tcp, stream_closed);
  uv_close((uv_handle_t *)&stream->output, stream_closed);
}

References platform_event_stream::active, event_write::arg, platform_event_stream::closing, event_write::done, ERROR_NETWORK, platform_event_stream::first, platform_event_stream::last, platform_event_stream::mutex, event_write::next, platform_event_stream::output, platform_event_stream::output_timer, platform_event_stream::queue_changed, platform_event_stream::queue_context, platform_event_stream::queued, platform_event_stream::reason, event_write::request, SAFE_FREE, event_write::size, event_write::storage, event_write::stream, platform_event_stream::tcp, platform_event_stream::timer, event_write::waiting, and platform_event_stream::writes.

Referenced by acip_async_tcp_accepted_create(), and platform_event_stream_adopt().

platform_event_stream_deadline()

asciichat_error_t platform_event_stream_deadline	(	platform_event_stream_t *	stream,
		uint64_t	timeout_ms )

Relative timeout converted to an absolute deadline; zero disables it.

Definition at line 357 of file event.c.

                                                                                                       {
  if (!stream || stream->closing)
    return SET_ERRNO(ERROR_INVALID_STATE, "Stream is closing");
  uv_timer_stop(&stream->timer);
  int status = timeout_ms ? uv_timer_start(&stream->timer, deadline_expired, timeout_ms, 0) : 0;
  return status < 0 ? event_error(status, "Set stream deadline") : ASCIICHAT_OK;
}

References ASCIICHAT_OK, platform_event_stream::closing, ERROR_INVALID_STATE, SET_ERRNO, event_write::stream, and platform_event_stream::timer.

platform_event_stream_observe_queue()

void platform_event_stream_observe_queue	(	platform_event_stream_t *	stream,
		platform_event_queue_fn	observer,
		void *	context )

Publish queued-byte changes without polling the stream. Configure before producers start. Invoked under the stream lock; only publish atomic state, never call stream/transport APIs. Context remains alive through on_close.

Definition at line 712 of file event.c.

                                                        {
  if (!stream)
    return;
  uv_mutex_lock(&stream->mutex);
  stream->queue_changed = observer;
  stream->queue_context = context;
  if (observer)
    observer(context, stream->queued);
  uv_mutex_unlock(&stream->mutex);
}

References platform_event_stream::mutex, platform_event_stream::queue_changed, platform_event_stream::queue_context, platform_event_stream::queued, and event_write::stream.

Referenced by acip_async_tcp_accepted_create(), acip_async_tcp_client_create(), acip_async_tcp_transport_create(), and acip_async_tcp_transport_detach().

platform_event_stream_pause()

void platform_event_stream_pause

(

platform_event_stream_t *

stream

)

Definition at line 341 of file event.c.

                                                                  {
  if (stream && !stream->closing)
    uv_read_stop((uv_stream_t *)&stream->tcp);
}

References platform_event_stream::closing, event_write::stream, and platform_event_stream::tcp.

Referenced by platform_event_stream_work().

platform_event_stream_peer()

asciichat_error_t platform_event_stream_peer	(	platform_event_stream_t *	stream,
		struct sockaddr_storage *	address,
		socklen_t *	length )

Definition at line 822 of file event.c.

                                                                {
  if (!stream || !address || !length)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid peer address output");
  int size = (int)*length;
  int status = uv_tcp_getpeername(&stream->tcp, (struct sockaddr *)address, &size);
  *length = (socklen_t)size;
  return status < 0 ? event_error(status, "Get stream peer") : ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, SET_ERRNO, and platform_event_stream::tcp.

platform_event_stream_queued_bytes()

size_t platform_event_stream_queued_bytes

(

const platform_event_stream_t *

stream

)

Definition at line 703 of file event.c.

                                                                                 {
  if (!stream)
    return 0;
  uv_mutex_lock((uv_mutex_t *)&stream->mutex);
  size_t bytes = stream->queued;
  uv_mutex_unlock((uv_mutex_t *)&stream->mutex);
  return bytes;
}

References platform_event_stream::mutex, platform_event_stream::queued, and event_write::stream.

platform_event_stream_request_close()

void platform_event_stream_request_close	(	platform_event_stream_t *	stream,
		asciichat_error_t	reason )

Reject new writes, drain accepted output, then close; write deadlines still apply.

Definition at line 625 of file event.c.

                                                                                                    {
  if (!stream)
    return;
  uv_mutex_lock(&stream->mutex);
  if (!stream->closing && !stream->close_requested) {
    stream->close_requested = true;
    stream->requested_reason = reason;
    uv_async_send(&stream->output);
  }
  uv_mutex_unlock(&stream->mutex);
}

References platform_event_stream::close_requested, platform_event_stream::closing, platform_event_stream::mutex, platform_event_stream::output, platform_event_stream::requested_reason, and event_write::stream.

platform_event_stream_resume()

asciichat_error_t platform_event_stream_resume

(

platform_event_stream_t *

stream

)

Definition at line 346 of file event.c.

                                                                                {
  if (!stream || stream->closing || !stream->read)
    return SET_ERRNO(ERROR_INVALID_STATE, "Stream cannot resume");
  int status = uv_read_start((uv_stream_t *)&stream->tcp, allocate_read, read_stream);
  return status < 0 ? event_error(status, "Start stream read") : ASCIICHAT_OK;
}

References ASCIICHAT_OK, platform_event_stream::closing, ERROR_INVALID_STATE, platform_event_stream::read, SET_ERRNO, event_write::stream, and platform_event_stream::tcp.

Referenced by platform_event_stream_start().

platform_event_stream_set_zerocopy()

void platform_event_stream_set_zerocopy	(	platform_event_stream_t *	stream,
		bool	enabled )

Configure copy avoidance before starting producers. The loop always owns I/O.

Definition at line 617 of file event.c.

                                                                                       {
  if (!stream)
    return;
  uv_mutex_lock(&stream->mutex);
  stream->zerocopy = enabled;
  uv_mutex_unlock(&stream->mutex);
}

References enabled, platform_event_stream::mutex, event_write::stream, and platform_event_stream::zerocopy.

platform_event_stream_start()

asciichat_error_t platform_event_stream_start	(	platform_event_stream_t *	stream,
		platform_event_read_fn	read,
		platform_event_close_fn	close,
		void *	arg )

Definition at line 326 of file event.c.

                                                                                        {
  if (!stream || stream->closing || !read)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid stream callbacks");
  // Retain cleanup callbacks even when configuring or starting reads fails.
  stream->read = read;
  stream->close = close;
  stream->arg = arg;
  int send_buffer = SOCKET_STREAM_SEND_BUFFER;
  int status = uv_send_buffer_size((uv_handle_t *)&stream->tcp, &send_buffer);
  if (status < 0)
    return event_error(status, "Bound stream send buffer");
  return platform_event_stream_resume(stream);
}

References platform_event_stream::arg, platform_event_stream::close, platform_event_stream::closing, ERROR_INVALID_PARAM, platform_event_stream_resume(), platform_event_stream::read, SET_ERRNO, event_write::stream, and platform_event_stream::tcp.

Referenced by acip_async_tcp_accepted_create(), and acip_async_tcp_client_create().

platform_event_stream_work()

asciichat_error_t platform_event_stream_work	(	platform_event_stream_t *	stream,
		platform_event_work_fn	run,
		platform_event_work_done_fn	done,
		void *	arg )

Run blocking application work outside the loop. One job per stream at a time; reads pause until done runs on the loop. Stream lifetime includes this job. write(), request_close(), and queued_bytes() are thread-safe while the caller holds a live stream reference; other stream operations are loop-only.

Definition at line 801 of file event.c.

                                                                                          {
  if (!stream || stream->closing || stream->working || !run || !done)
    return SET_ERRNO(ERROR_INVALID_STATE, "Stream work already active or closing");
  event_work_t *work = SAFE_CALLOC(1, sizeof(*work), event_work_t *);
  work->stream = stream;
  work->run = run;
  work->done = done;
  work->arg = arg;
  work->work.data = work;
  platform_event_stream_pause(stream);
  int status = uv_queue_work(stream->tcp.loop, &work->work, run_work, finish_work);
  if (status < 0) {
    SAFE_FREE(work);
    return event_error(status, "Queue stream work");
  }
  stream->working = true;
  ++stream->handles;
  return ASCIICHAT_OK;
}

References event_work_t::arg, ASCIICHAT_OK, platform_event_stream::closing, event_work_t::done, ERROR_INVALID_STATE, platform_event_stream::handles, platform_event_stream_pause(), event_work_t::run, SAFE_CALLOC, SAFE_FREE, SET_ERRNO, event_work_t::stream, platform_event_stream::tcp, event_work_t::work, and platform_event_stream::working.

platform_event_stream_write()

asciichat_error_t platform_event_stream_write	(	platform_event_stream_t *	stream,
		const socket_buffer_t *	buffers,
		size_t	count,
		platform_event_write_fn	done,
		void *	arg )

Copies input before returning. Completion means local write completion, not peer acknowledgement. Rejected writes never invoke completion. Accepted writes invoke completion exactly once, including during shutdown, before on_close.

Definition at line 637 of file event.c.

                                                                                                     {
  if (!stream || !buffers || !count || count > SOCKET_IOV_MAX)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid asynchronous write");
  size_t size = 0;
  for (size_t i = 0; i < count; ++i) {
    if ((!buffers[i].data && buffers[i].len) || buffers[i].len > EVENT_WRITE_LIMIT - size)
      return SET_ERRNO(ERROR_NETWORK_SIZE, "Asynchronous write too large");
    size += buffers[i].len;
  }
  uv_mutex_lock(&stream->mutex);
  if (stream->closing || stream->close_requested || !size || stream->writes >= EVENT_WRITE_COUNT_LIMIT ||
      size > EVENT_WRITE_LIMIT - stream->queued) {
    uv_mutex_unlock(&stream->mutex);
    return SET_ERRNO(ERROR_NETWORK_SIZE, "Asynchronous output queue full");
  }
  bool mapped = stream->zerocopy && size >= SOCKET_ZEROCOPY_THRESHOLD;
  event_write_t **cached = &stream->recycled;
  while (*cached && (mapped || (*cached)->storage.capacity < size))
    cached = &(*cached)->next;
  event_write_t *write = *cached;
  socket_send_buffer_t storage = {0};
  if (write) {
    *cached = write->next;
    --stream->recycled_count;
    storage = write->storage;
  } else {
    write = SAFE_MALLOC(sizeof(*write), event_write_t *);
  }
  *write = (event_write_t){.stream = stream, .done = done, .arg = arg, .size = size, .storage = storage};
  if (!write->storage.data && mapped) {
    if (socket_send_buffer_alloc(size, &write->storage) != ASCIICHAT_OK) {
      SAFE_FREE(write);
      uv_mutex_unlock(&stream->mutex);
      return SET_ERRNO(ERROR_MEMORY, "Async copy avoidance exceeds process buffer budget");
    }
  } else if (!write->storage.data) {
    write->storage.data = buffer_alloc(NULL, size);
    write->storage.capacity = size;
  }
  if (!write->storage.data) {
    SAFE_FREE(write);
    uv_mutex_unlock(&stream->mutex);
    return SET_ERRNO(ERROR_MEMORY, "Async output exceeds process buffer budget");
  }
  write->request.data = write;
  size_t offset = 0;
  for (size_t i = 0; i < count; ++i) {
    if (buffers[i].len)
      memcpy((char *)write->storage.data + offset, buffers[i].data, buffers[i].len);
    offset += buffers[i].len;
  }
  stream->queued += size;
  if (stream->queue_changed)
    stream->queue_changed(stream->queue_context, stream->queued);
  ++stream->writes;
  if (stream->last)
    stream->last->next = write;
  else
    stream->first = write;
  stream->last = write;
  uv_async_send(&stream->output);
  uv_mutex_unlock(&stream->mutex);
  return ASCIICHAT_OK;
}

References platform_event_stream::arg, ASCIICHAT_OK, buffer_alloc(), platform_event_stream::close_requested, platform_event_stream::closing, event_write::done, ERROR_INVALID_PARAM, ERROR_MEMORY, ERROR_NETWORK_SIZE, EVENT_WRITE_COUNT_LIMIT, EVENT_WRITE_LIMIT, platform_event_stream::first, platform_event_stream::last, platform_event_stream::mutex, event_write::next, event_write::offset, platform_event_stream::output, platform_event_stream::queue_changed, platform_event_stream::queue_context, platform_event_stream::queued, platform_event_stream::recycled, platform_event_stream::recycled_count, event_write::request, SAFE_FREE, SAFE_MALLOC, SET_ERRNO, event_write::size, event_write::storage, event_write::stream, platform_event_stream::writes, and platform_event_stream::zerocopy.