test_uv.c

#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sysexits.h>
#include <uv.h>

#define NTHREADS 4

#define UV_CHK(A)	do { if ((A) != 0) { printf("%s:%d, %s, %s\n", __FILE__, __LINE__, uv_err_name(A), uv_strerror(A)); abort(); } } while (false);

struct dnsb_dwrk
{
	uv_thread_t id;
	uv_loop_t* loop;
	uv_os_sock_t os_recv_socket;
	uv_udp_t recv_socket;
	struct sockaddr_in recv_addr;
	uv_async_t async;
	uv_signal_t sigint;
	uv_signal_t sigterm;
};

struct dnsb_req
{
	uv_udp_t* handle;
	char* data;
	size_t size;
	struct sockaddr addr;
};

static void dnsb_dwrk_int_term_sighandler(uv_signal_t* _handle, int _signum)
{
	printf("[%p] Got %d signal, sending async\n", (void*)_handle->loop, _signum);

	uv_async_send((uv_async_t*)_handle->data);
	uv_signal_stop(_handle);

	return;
}

static void dnsb_dwrk_on_send(uv_udp_send_t* _req, int _status)
{
	(void)_status;

	// TODO: _req->data SHOULD point to current buf, no?
	printf("%p\n", (void*)_req->bufs); // TODO: this is also NULL and this is insanity
	// _req->nbufs is NOT NULL!
	for (size_t i = 0; i < _req->nbufs; i++)
	{
		if (&_req->bufs[i])
		{
			printf("Freeing %zu/%u buf, buf is %p\n", i, _req->nbufs, (void*)&_req->bufs[i]);
			printf("Freeing %zu/%u buf, base is %p\n", i, _req->nbufs, (void*)_req->bufs[i].base);
			free(_req->bufs[i].base);
			free(&_req->bufs[i]);
		}
	}

	free(_req);

	return;
}

static void dnsb_dwrk_alloc_buf(uv_handle_t* _handle, size_t _suggested_size, uv_buf_t* _buf)
{
	(void)_handle;

	_buf->base = malloc(_suggested_size);
	_buf->len = _suggested_size;

	return;
}

static void dnsb_wrk(uv_work_t* _work)
{
	struct dnsb_req* data = _work->data;

	printf("Loop %p got: ", (void*)_work->loop);
	for (size_t i = 0; i < data->size; i++)
		printf("%#02x ", data->data[i]);
	printf("\n");

	uv_udp_send_t* rep = malloc(sizeof(uv_udp_send_t));
	uv_buf_t* buf = malloc(sizeof(uv_buf_t));
	dnsb_dwrk_alloc_buf(NULL, 6, buf);
	buf->base[0] = 'h';
	buf->base[1] = 'e';
	buf->base[2] = 'l';
	buf->base[3] = 'l';
	buf->base[4] = 'o';
	buf->base[5] = '\0';

	uv_udp_send(rep, data->handle, buf, 1, (const struct sockaddr*)&data->addr, dnsb_dwrk_on_send);

	return;
}

void dnsb_wrk_done(uv_work_t* _work, int _status)
{
	(void)_status;
	struct dnsb_req* data = _work->data;

	free(data->data);
	free(data);
	free(_work);

	return;
}

static void dnsb_dwrk_on_read(uv_udp_t* _req, ssize_t _nread, const uv_buf_t* _buf, const struct sockaddr* _addr, unsigned _flags)
{
	(void)_addr;
	(void)_flags;
	uv_work_t* work = NULL;
	struct dnsb_req* req = NULL;

	// should be error
	if (_nread < 0)
	{
		uv_close((uv_handle_t*)_req, NULL);
		free(_buf->base);
		return;
	}

	// empty packet
	if (_nread == 0 || !_addr)
	{
		free(_buf->base);
		return;
	}

	req = malloc(sizeof(struct dnsb_req));
	req->handle = _req;
	req->data = _buf->base;
	req->size = (size_t)_nread;
	req->addr = *_addr;
	work = malloc(sizeof(uv_work_t));
	work->data = (void*)req;
	uv_queue_work(_req->loop, work, dnsb_wrk, dnsb_wrk_done);

	return;
}

static void dns_dwrk_async_handler(uv_async_t* _handle)
{
	printf("[%p] Got async, stopping loop\n", (void*)_handle->loop);

	uv_stop(_handle->loop);

	return;
}

static void dnsb_dwrk(void* _data)
{
	int option = 1;

	struct dnsb_dwrk* data = _data;

	data->loop = malloc(sizeof(uv_loop_t));
	uv_loop_init(data->loop);

	uv_async_init(data->loop, &data->async, dns_dwrk_async_handler);

	uv_signal_init(data->loop, &data->sigint);
	uv_signal_init(data->loop, &data->sigterm);
	data->sigint.data = &data->async;
	data->sigterm.data = &data->async;
	uv_signal_start(&data->sigint, dnsb_dwrk_int_term_sighandler, SIGINT);
	uv_signal_start(&data->sigterm, dnsb_dwrk_int_term_sighandler, SIGTERM);

	data->os_recv_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
	setsockopt(data->os_recv_socket, SOL_SOCKET, SO_REUSEPORT, (const void*)&option, sizeof(option));
	uv_udp_init(data->loop, &data->recv_socket);
	uv_udp_open(&data->recv_socket, data->os_recv_socket);
	uv_ip4_addr("127.0.0.1", 53535, &data->recv_addr);
	uv_udp_bind(&data->recv_socket, (const struct sockaddr*)&data->recv_addr, UV_UDP_REUSEADDR);
	uv_udp_recv_start(&data->recv_socket, dnsb_dwrk_alloc_buf, dnsb_dwrk_on_read);

	printf("Running loop: %p\n", (void*)data->loop);
	uv_run(data->loop, UV_RUN_DEFAULT);

	printf("Exiting loop: %p\n", (void*)data->loop);
	uv_udp_recv_stop(&data->recv_socket);
	uv_loop_close(data->loop);
	free(data->loop);

	return;
}

int main(int _argc, char** _argv)
{
	(void)_argc;
	(void)_argv;

	struct dnsb_dwrk* dnsb_dwrks = malloc(NTHREADS * sizeof(struct dnsb_dwrk));

	for (size_t i = 0; i < NTHREADS; i++)
	{
		uv_thread_create(&dnsb_dwrks[i].id, dnsb_dwrk, (void*)&dnsb_dwrks[i]);
	}

	for (size_t i = 0; i < NTHREADS; i++)
	{
		uv_thread_join(&dnsb_dwrks[i].id);
	}

	free(dnsb_dwrks);

	printf("Ciao\n");

	exit(EX_OK);
}