quentin/tor_multipath_voip
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Compiling solution

Browse source
This commit is contained in:
Quentin 2019-02-18 21:55:53 +01:00
parent 09d4c24857
commit 5e291f21a7
10 changed files with 222 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

343
src/algo_naive.c
View file

@ -2,14 +2,27 @@
struct naive_ctx { struct naive_ctx {
int ref_count; int ref_count;
GHashTable* packet_buffer; struct buffer_packet bps[10];
GQueue* free_buffer; // Available buffers
GHashTable* used_buffer; // Buffers used for reading or writing
GQueue* read_waiting; // Who wait to be notified for a read
GHashTable* write_waiting; // Structure to track packets waiting to be written
}; };
void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
void free_nothing(void* app_ctx) {} void free_nothing(void* app_ctx) {}
void free_naive(void* app_ctx) { void free_naive(void* app_ctx) {
struct naive_ctx* ctx = (struct naive_ctx*) app_ctx; struct naive_ctx* ctx = (struct naive_ctx*) app_ctx;
ctx->ref_count--; ctx->ref_count--;
if (ctx->ref_count <= 0) free(ctx); if (ctx->ref_count > 0) return;
g_queue_free(ctx->free_buffer);
g_queue_free(ctx->read_waiting);
g_hash_table_destroy (ctx->used_buffer);
free(ctx);
} }
char* get_port(char* out, char* in) { char* get_port(char* out, char* in) {
@ -17,39 +30,35 @@ char* get_port(char* out, char* in) {
return out; return out;
} }
void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
int conn_sock1, conn_sock2, port; int conn_sock1, conn_sock2;
struct sockaddr_in addr; struct sockaddr_in addr;
socklen_t in_len; socklen_t in_len;
char url[1024]; char url[1024], port[6];
struct evt_core_cat local_cat = {0}; struct evt_core_cat local_cat = {0};
struct evt_core_fdinfo fdinfo = {0}; struct evt_core_fdinfo to_fdinfo = {0};
fdinfo.cat = &local_cat; to_fdinfo.cat = &local_cat;
fdinfo.url = url; to_fdinfo.url = url;
in_len = sizeof(addr); in_len = sizeof(addr);
conn_sock1 = accept(fd, (struct sockaddr*)&addr, &in_len); conn_sock1 = accept(fdinfo->fd, (struct sockaddr*)&addr, &in_len);
if (conn_sock1 == -1) goto co_error; if (conn_sock1 == -1) goto co_error;
conn_sock2 = dup(conn_sock1); conn_sock2 = dup(conn_sock1);
if (conn_sock2 == -1) goto co_error; if (conn_sock2 == -1) goto co_error;
//printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2); //printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2);
struct evt_core_fdinfo* listen_info = g_hash_table_lookup(ctx->socklist, &fd); get_port(port, fdinfo->url);
if (listen_info == NULL) goto co_error;
sscanf(listen_info->url, "tcp:listen:127.0.0.1:%d", &port);
fdinfo.fd = conn_sock1; to_fdinfo.fd = conn_sock1;
fdinfo.cat->name = "tcp-read"; to_fdinfo.cat->name = "tcp-read";
sprintf(fdinfo.url, "tcp:read:127.0.0.1:%d", port); sprintf(to_fdinfo.url, "tcp:read:127.0.0.1:%s", port);
evt_core_add_fd (ctx, &fdinfo); evt_core_add_fd (ctx, &to_fdinfo);
fdinfo.fd = conn_sock2; to_fdinfo.fd = conn_sock2;
fdinfo.cat->name = "tcp-write"; to_fdinfo.cat->name = "tcp-write";
sprintf(fdinfo.url, "tcp:write:127.0.0.1:%d", port); sprintf(to_fdinfo.url, "tcp:write:127.0.0.1:%s", port);
evt_core_add_fd (ctx, &fdinfo); evt_core_add_fd (ctx, &to_fdinfo);
printf("Selected port: %d\n", port);
return; return;
@ -58,59 +67,115 @@ co_error:
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
struct buffer_packet* get_bp(struct evt_core_cat* cat, int fd, int is_read_buffer) { /**
* Returns a buffer if available, NULL otherwise
*/
struct buffer_packet* get_read_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
struct buffer_packet* bp; struct buffer_packet* bp;
struct naive_ctx* ctx = cat->app_ctx;
bp = g_hash_table_lookup (ctx->packet_buffer, &fd); // 1. Check if we don't have a buffer
bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd);
if (bp != NULL) return bp; if (bp != NULL) return bp;
if (!is_read_buffer) goto alloc_error;
bp = malloc(sizeof(struct buffer_packet)); // 2. Get a new buffer otherwise
if (bp == NULL) goto alloc_error; bp = g_queue_pop_head(app_ctx->free_buffer);
memset(bp, 0, sizeof(struct buffer_packet)); if (bp == NULL) {
bp->fdread = fd; // 2.1 If no buffer is available, we subscribe to be notified later
bp->fdwrite = -1; g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd));
bp->ref_count++; return NULL;
g_hash_table_insert(ctx->packet_buffer, &(bp->fdread), bp); }
// 3. Update state
g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp);
return bp; return bp;
alloc_error:
perror("alloc error");
exit(EXIT_FAILURE);
} }
void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); /**
void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); * Returns a buffer if available, NULL otherwise
void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); */
void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); struct buffer_packet* get_write_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
struct buffer_packet* bp;
GQueue* q;
void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { // 1. Check if we don't have a buffer
bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd);
if (bp != NULL) return bp;
// 2. Check our waiting queue otherwise
if ((q = g_hash_table_lookup(app_ctx->write_waiting, &(fdinfo->fd))) == NULL) return NULL;
bp = g_queue_pop_head(q);
if (bp == NULL) return NULL; // No packet to process
// 3. Update state
g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp);
return bp;
}
void mv_buffer_rtow(struct naive_ctx* app_ctx,
struct evt_core_fdinfo* from,
struct evt_core_fdinfo* to,
struct buffer_packet* bp) {
// 1. We get the target writing queue
GQueue* q;
q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd));
if (q == NULL) {
q = g_queue_new ();
g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q);
}
// 2. We move the buffer to the target queue
g_hash_table_remove(app_ctx->used_buffer, &from->fd);
g_queue_push_tail(q, bp);
}
void mv_buffer_wtor(struct naive_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
g_queue_push_tail (app_ctx->free_buffer, bp);
g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd));
}
void notify_read(struct evt_core_ctx* ctx, struct naive_ctx* app_ctx) {
struct evt_core_fdinfo* next_fdinfo = NULL;
while (next_fdinfo == NULL) {
int fd = GPOINTER_TO_INT(g_queue_pop_head(app_ctx->read_waiting));
if (fd == 0) break;
next_fdinfo = evt_core_get_from_fd (ctx, fd);
if (strcmp(next_fdinfo->cat->name, "tcp-read") == 0) on_tcp_read(ctx, next_fdinfo);
else if (strcmp(next_fdinfo->cat->name, "udp-read") == 0) on_udp_read(ctx, next_fdinfo);
else {
fprintf(stderr, "A fd from category %s can't be stored in read_waiting\n", next_fdinfo->cat->name);
exit(EXIT_FAILURE);
}
}
}
void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp; struct buffer_packet* bp;
struct evt_core_fdinfo *to_fdinfo = NULL; struct evt_core_fdinfo *to_fdinfo = NULL;
struct naive_ctx* app_ctx = cat->app_ctx; struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
int read_res = FDS_READY; int read_res = FDS_READY;
char url[255]; char url[255];
// 1. Read in our buffer packet // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later
bp = get_bp(cat, fd, TRUE); if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return;
// 2. Try to read a whole packet in the buffer
while (read_res != FDS_AGAIN && bp->mode == BP_READING) { while (read_res != FDS_AGAIN && bp->mode == BP_READING) {
read_res = read_packet_from_tcp (fd, bp); read_res = read_packet_from_tcp (fdinfo->fd, bp);
if (read_res == FDS_ERR) goto co_error; if (read_res == FDS_ERR) goto co_error;
} }
if (bp->mode != BP_SWITCH_WRITE) return; if (bp->mode != BP_WRITING) return;
// 2. Packet has just been read, choose a destination // 3. A whole packet has been read, we will find someone to write it
sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port);
to_fdinfo = evt_core_get_from_url (ctx, url); to_fdinfo = evt_core_get_from_url (ctx, url);
if (to_fdinfo == NULL) goto co_error; if (to_fdinfo == NULL) goto co_error;
// 3. Configure packet buffer for the destination // 4. We move the buffer and notify the target
bp->mode = BP_WRITING; mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
bp->fdwrite = to_fdinfo->fd; on_udp_write(ctx, to_fdinfo);
bp->ref_count++;
g_hash_table_insert(app_ctx->packet_buffer, &(bp->fdwrite), bp);
on_udp_write(ctx, to_fdinfo->cat, to_fdinfo->fd);
return; return;
co_error: co_error:
@ -118,140 +183,136 @@ co_error:
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp; struct buffer_packet* bp;
struct evt_core_fdinfo *to_fdinfo; struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
int write_res = FDS_READY; int write_res = FDS_READY;
char url[255];
bp = get_bp(cat, fd, FALSE); // 1. Get current write buffer OR a buffer from the waiting queue OR leave
if (bp == NULL) return; if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return;
// 2. Write data from the buffer to the socket
while (write_res != FDS_AGAIN && bp->mode == BP_WRITING) { while (write_res != FDS_AGAIN && bp->mode == BP_WRITING) {
write_res = write_packet_to_tcp(fd, bp); write_res = write_packet_to_tcp(fdinfo->fd, bp);
if (write_res == FDS_ERR) goto co_error; if (write_res == FDS_ERR) goto co_error;
} }
if (bp->mode != BP_READING) return;
// 3. A whole packet has been written
// Release the buffer and notify
mv_buffer_wtor(app_ctx, fdinfo, bp);
notify_read(ctx, app_ctx);
return;
co_error: co_error:
perror("Failed to TCP write"); perror("Failed to TCP write");
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp;
} struct evt_core_fdinfo *to_fdinfo;
struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { int read_res = FDS_READY;
}
void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
// Init data structures for the transfer
char url[255]; char url[255];
struct naive_ctx* app_ctx = cat->app_ctx;
struct buffer_packet* bp = &(app_ctx->tcp_to_udp);
struct evt_core_fdinfo* fdinfo;
int read_res = FDS_READY;
int write_res = FDS_READY;
// Either we can read something, either we can write something // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later
while ((read_res != FDS_AGAIN && bp->mode == BP_READING) if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return;
|| (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) {
// 1. Read packet from TCP socket // 2. Read packet from socket
if (bp->mode == BP_READING) { read_res = read_packet_from_udp (fdinfo->fd, bp);
read_res = read_packet_from_tcp (fd, bp); if (read_res == FDS_ERR) goto co_error;
if (read_res == FDS_ERR) goto co_error; if (bp->mode != BP_WRITING) return;
}
// 2. Write packet to UDP socket // 3. A whole packet has been read, we will find someone to write it
if (bp->mode == BP_WRITING) { sprintf(url, "tcp:write:127.0.0.1:7500");
sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); to_fdinfo = evt_core_get_from_url (ctx, url);
fdinfo = evt_core_get_from_url(ctx, url); if (to_fdinfo == NULL) goto co_error;
if (fdinfo == NULL) goto co_error;
write_res = write_packet_to_udp(fdinfo->fd, bp); // 4. We move the buffer and notify the target
if (read_res == FDS_ERR) goto co_error; mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
} on_tcp_write(ctx, to_fdinfo);
}
return;
co_error: co_error:
perror("Failed to handle read write for tcp_to_udp"); perror("Failed to UDP read");
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
void udp_to_tcp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
// Get target file descriptor struct buffer_packet* bp;
struct evt_core_cat* tcp = g_hash_table_lookup (ctx->catlist, "tcp-write"); struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
if (tcp == NULL || tcp->socklist->len < 1) goto co_error;
int tcp_fd = g_array_index(tcp->socklist, int, 0);
// Init data structures for the transfer
struct naive_ctx* app_ctx = cat->app_ctx;
struct buffer_packet* bp = &(app_ctx->udp_to_tcp);
int read_res = FDS_READY;
int write_res = FDS_READY; int write_res = FDS_READY;
// Either we can read something, either we can write something // 1. Get current write buffer OR a buffer from the waiting queue OR leave
while ((read_res != FDS_AGAIN && bp->mode == BP_READING) if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return;
|| (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) {
// 1. Read packet from UDP socket // 2. Write buffer
if (bp->mode == BP_READING) { write_res = write_packet_to_udp(fdinfo->fd, bp);
read_res = read_packet_from_udp (fd, bp); if (write_res == FDS_ERR) goto co_error;
if (read_res == FDS_ERR) goto co_error; if (bp->mode != BP_READING) return;
}
// 2. Write packet to TCP socket // 3. A whole packet has been written
if (bp->mode == BP_WRITING) { // Release the buffer and notify
write_res = write_packet_to_tcp(tcp_fd, bp); mv_buffer_wtor(app_ctx, fdinfo, bp);
if (write_res == FDS_ERR) goto co_error; notify_read(ctx, app_ctx);
}
}
return;
co_error: co_error:
perror("Failed to handle read write for udp_to_tcp"); perror("Failed to UDP write");
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
void free_buffer_packet(void *v) {
struct buffer_packet* bp = (struct buffer_packet*) v;
bp->ref_count--;
if (bp->ref_count == 0) free(v);
}
void free_simple(void* v) { void naive_free_simple(void* v) {
free(v); free(v);
} }
void on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
struct buffer_packet* bp;
// 1. If has a "used" buffer, remove it
bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd));
if (bp != NULL) {
g_hash_table_remove (app_ctx->used_buffer, &(fdinfo->fd));
memset(bp, 0, sizeof(struct buffer_packet));
g_queue_push_tail(app_ctx->free_buffer, bp);
}
// 2. If appears in the write waiting queue, remove it
GQueue* writew = g_hash_table_lookup (app_ctx->write_waiting, &(fdinfo->fd));
while (writew != NULL && (bp = g_queue_pop_head (writew)) != NULL) {
memset(bp, 0, sizeof(struct buffer_packet));
g_queue_push_tail(app_ctx->free_buffer, bp);
}
g_hash_table_remove (app_ctx->write_waiting, &(fdinfo->fd));
// 3. If appears in the read waiting queue, remove it
g_queue_remove_all (app_ctx->read_waiting, &(fdinfo->fd));
}
void algo_naive(struct algo_skel* as) { void algo_naive(struct algo_skel* as) {
struct naive_ctx* ctx = malloc(sizeof(struct naive_ctx)); struct naive_ctx* ctx = malloc(sizeof(struct naive_ctx));
if (ctx == NULL) goto init_err; if (ctx == NULL) goto init_err;
memset(ctx, 0, sizeof(struct naive_ctx)); memset(ctx, 0, sizeof(struct naive_ctx));
ctx->packet_buffer = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, free_buffer_packet); ctx->free_buffer = g_queue_new ();
ctx->read_waiting = g_queue_new ();
as->on_init.name = "init"; ctx->used_buffer = g_hash_table_new(g_int_hash, g_int_equal);
as->on_init.flags = 0; ctx->write_waiting = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, naive_free_simple);
as->on_init.app_ctx = NULL; for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) {
as->on_init.free_app_ctx = NULL; g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i]));
as->on_init.cb = NULL; }
as->on_init.socklist = NULL;
as->on_tcp_co.name = "tcp-listen"; as->on_tcp_co.name = "tcp-listen";
as->on_tcp_co.flags = EPOLLIN; as->on_tcp_co.flags = EPOLLIN;
as->on_tcp_co.app_ctx = NULL;
as->on_tcp_co.free_app_ctx = free_nothing; as->on_tcp_co.free_app_ctx = free_nothing;
as->on_tcp_co.cb = on_tcp_co; as->on_tcp_co.cb = on_tcp_co;
as->on_tcp_co.socklist = NULL;
as->on_tcp_read.name = "tcp-read"; as->on_tcp_read.name = "tcp-read";
as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP; as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP;
as->on_tcp_read.app_ctx = ctx; as->on_tcp_read.app_ctx = ctx;
as->on_tcp_read.free_app_ctx = free_naive; as->on_tcp_read.free_app_ctx = free_naive;
as->on_tcp_read.cb = on_tcp_read; as->on_tcp_read.cb = on_tcp_read;
as->on_tcp_read.socklist = NULL; as->on_tcp_read.err_cb = on_err;
ctx->ref_count++; ctx->ref_count++;
as->on_udp_read.name = "udp-read"; as->on_udp_read.name = "udp-read";
@ -259,7 +320,7 @@ void algo_naive(struct algo_skel* as) {
as->on_udp_read.app_ctx = ctx; as->on_udp_read.app_ctx = ctx;
as->on_udp_read.free_app_ctx = free_naive; as->on_udp_read.free_app_ctx = free_naive;
as->on_udp_read.cb = on_udp_read; as->on_udp_read.cb = on_udp_read;
as->on_udp_read.socklist = NULL; as->on_udp_read.err_cb = on_err;
ctx->ref_count++; ctx->ref_count++;
as->on_tcp_write.name = "tcp-write"; as->on_tcp_write.name = "tcp-write";
@ -267,7 +328,7 @@ void algo_naive(struct algo_skel* as) {
as->on_tcp_write.app_ctx = ctx; as->on_tcp_write.app_ctx = ctx;
as->on_tcp_write.free_app_ctx = free_naive; as->on_tcp_write.free_app_ctx = free_naive;
as->on_tcp_write.cb = on_tcp_write; as->on_tcp_write.cb = on_tcp_write;
as->on_tcp_write.socklist = NULL; as->on_tcp_write.err_cb = on_err;
ctx->ref_count++; ctx->ref_count++;
as->on_udp_write.name = "udp-write"; as->on_udp_write.name = "udp-write";
@ -275,7 +336,7 @@ void algo_naive(struct algo_skel* as) {
as->on_udp_write.app_ctx = ctx; as->on_udp_write.app_ctx = ctx;
as->on_udp_write.free_app_ctx = free_naive; as->on_udp_write.free_app_ctx = free_naive;
as->on_udp_write.cb = on_udp_write; as->on_udp_write.cb = on_udp_write;
as->on_udp_write.socklist = NULL; as->on_udp_write.err_cb = on_err;
ctx->ref_count++; ctx->ref_count++;
return; return;

1
src/algo_skel.h
View file

@ -8,7 +8,6 @@
#include "utils.h" #include "utils.h"
struct algo_skel { struct algo_skel {
struct evt_core_cat on_init;
struct evt_core_cat on_udp_read; struct evt_core_cat on_udp_read;
struct evt_core_cat on_tcp_read; struct evt_core_cat on_tcp_read;
struct evt_core_cat on_udp_write; struct evt_core_cat on_udp_write;

2
src/donar.c
View file

@ -74,8 +74,8 @@ in_error:
terminate: terminate:
if (onion_file != NULL) free(onion_file); if (onion_file != NULL) free(onion_file);
if (port != NULL) free(port);
if (algo != NULL) free(algo); if (algo != NULL) free(algo);
g_ptr_array_free(ports, TRUE);
return errored; return errored;
} }

1
src/donar_client.c
View file

@ -150,7 +150,6 @@ void donar_client(struct donar_client_ctx* ctx, struct algo_skel* algo, char* on
.socklist = NULL .socklist = NULL
}; };
evt_core_add_cat (&(ctx->evts), &init_socks5); evt_core_add_cat (&(ctx->evts), &init_socks5);
evt_core_add_cat (&(ctx->evts), &(algo->on_init));
evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co));
evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read));
evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read));

1
src/donar_server.c
View file

@ -82,7 +82,6 @@ socket_failed:
void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo, GPtrArray* ports) { void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo, GPtrArray* ports) {
evt_core_init (&(ctx->evts)); evt_core_init (&(ctx->evts));
evt_core_add_cat (&(ctx->evts), &(algo->on_init));
evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co));
evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read));
evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read));

21
src/evt_core.c
View file

@ -52,6 +52,7 @@ void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat) {
dyn->cb = cat->cb; dyn->cb = cat->cb;
dyn->name = strdup(cat->name); dyn->name = strdup(cat->name);
dyn->flags = cat->flags; dyn->flags = cat->flags;
dyn->err_cb = cat->err_cb;
dyn->socklist = g_array_new (FALSE, FALSE, sizeof(struct evt_core_fdinfo*)); dyn->socklist = g_array_new (FALSE, FALSE, sizeof(struct evt_core_fdinfo*));
if (dyn->name == NULL) { if (dyn->name == NULL) {
@ -133,11 +134,6 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
struct evt_core_fdinfo* fdinfo; struct evt_core_fdinfo* fdinfo;
struct evt_core_cat* cat; struct evt_core_cat* cat;
cat = g_hash_table_lookup(ctx->catlist, "init");
if (cat != NULL) {
cat->cb(ctx, cat, -1);
}
printf("--- Start main loop\n"); printf("--- Start main loop\n");
int num_fd, n = 0; int num_fd, n = 0;
while(1) { while(1) {
@ -156,9 +152,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
if (evt & EPOLLHUP) fprintf(stderr, "Epoll Hup Event. "); if (evt & EPOLLHUP) fprintf(stderr, "Epoll Hup Event. ");
if (evt & EPOLLERR) fprintf(stderr, "Epoll Err Event. "); if (evt & EPOLLERR) fprintf(stderr, "Epoll Err Event. ");
cat = evt_core_rm_fd (ctx, err_fd); fdinfo = evt_core_get_from_fd(ctx, err_fd);
if (cat != NULL) { if (fdinfo != NULL) {
fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, cat->name); fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, fdinfo->cat->name);
if (fdinfo->cat->err_cb != NULL) fdinfo->cat->err_cb(ctx, fdinfo);
evt_core_rm_fd (ctx, err_fd);
} else { } else {
fprintf(stderr, "The file descriptor is not registered in a category, this is probably a logic error\n"); fprintf(stderr, "The file descriptor is not registered in a category, this is probably a logic error\n");
close (err_fd); close (err_fd);
@ -172,16 +170,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
fprintf(stderr, "Ignoring file descriptor %d as it is not registered. This is a bug.\n", events[n].data.fd); fprintf(stderr, "Ignoring file descriptor %d as it is not registered. This is a bug.\n", events[n].data.fd);
continue; continue;
} }
fdinfo->cat->cb(ctx, fdinfo->cat, events[n].data.fd); fdinfo->cat->cb(ctx, fdinfo);
} }
} }
cat = g_hash_table_lookup(ctx->catlist, "destroy");
if (cat != NULL) {
cat->cb(ctx, cat, -1);
}
evt_core_free(ctx); evt_core_free(ctx);
} }

4
src/evt_core.h
View file

@ -12,14 +12,16 @@
struct evt_core_ctx; struct evt_core_ctx;
struct evt_core_cat; struct evt_core_cat;
struct evt_core_fdinfo;
typedef void (*evt_core_free_app_ctx)(void*); typedef void (*evt_core_free_app_ctx)(void*);
typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_cat*, int fd); typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_fdinfo*);
struct evt_core_cat { struct evt_core_cat {
void* app_ctx; void* app_ctx;
evt_core_free_app_ctx free_app_ctx; evt_core_free_app_ctx free_app_ctx;
evt_core_cb cb; evt_core_cb cb;
evt_core_cb err_cb;
char* name; char* name;
int flags; int flags;
GArray* socklist; GArray* socklist;

6
src/meas_lat.c
View file

@ -7,10 +7,10 @@ struct timer_ctx {
uint64_t counter; uint64_t counter;
}; };
void on_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_udp(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
} }
void on_timer(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { void on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
} }
int main(int argc, char** argv) { int main(int argc, char** argv) {
@ -23,6 +23,7 @@ int main(int argc, char** argv) {
.app_ctx = NULL, .app_ctx = NULL,
.free_app_ctx = NULL, .free_app_ctx = NULL,
.cb = on_udp, .cb = on_udp,
.err_cb = NULL,
.name = "udp-read", .name = "udp-read",
.flags = EPOLLIN | EPOLLET, .flags = EPOLLIN | EPOLLET,
.socklist = NULL .socklist = NULL
@ -31,6 +32,7 @@ int main(int argc, char** argv) {
.app_ctx = &apptime, .app_ctx = &apptime,
.free_app_ctx = NULL, .free_app_ctx = NULL,
.cb = on_timer, .cb = on_timer,
.err_cb = NULL,
.name = "timer", .name = "timer",
.flags = EPOLLIN | EPOLLET, .flags = EPOLLIN | EPOLLET,
.socklist = NULL .socklist = NULL

8
src/packet.c
View file

@ -21,7 +21,7 @@ enum FD_STATE read_packet_from_tcp(int fd, struct buffer_packet* bp) {
bp->aread += nread; bp->aread += nread;
} }
bp->mode = BP_SWITCH_WRITE; bp->mode = BP_WRITING;
bp->awrite = 0; bp->awrite = 0;
return FDS_READY; return FDS_READY;
@ -38,7 +38,7 @@ enum FD_STATE write_packet_to_tcp(int fd, struct buffer_packet* bp) {
bp->awrite += nwrite; bp->awrite += nwrite;
} }
bp->mode = BP_SWITCH_READ; bp->mode = BP_READING;
bp->aread = 0; bp->aread = 0;
return FDS_READY; return FDS_READY;
@ -60,7 +60,7 @@ enum FD_STATE write_packet_to_udp(int fd, struct buffer_packet* bp) {
if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN; if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN;
if (nwrite != bytes_to_send) return FDS_ERR; if (nwrite != bytes_to_send) return FDS_ERR;
bp->mode = BP_SWITCH_READ; bp->mode = BP_READING;
bp->aread = 0; bp->aread = 0;
return FDS_READY; return FDS_READY;
@ -81,7 +81,7 @@ enum FD_STATE read_packet_from_udp (int fd, struct buffer_packet* bp) {
bp->ip.ap.str.size = nread + pkt_header_size; bp->ip.ap.str.size = nread + pkt_header_size;
bp->mode = BP_SWITCH_WRITE; bp->mode = BP_WRITING;
bp->awrite = 0; bp->awrite = 0;
return FDS_READY; return FDS_READY;

7
src/packet.h
View file

@ -24,9 +24,7 @@ enum FD_STATE {
enum BP_MODE { enum BP_MODE {
BP_READING, BP_READING,
BP_WRITING, BP_WRITING
BP_SWITCH_WRITE,
BP_SWITCH_READ
}; };
union abstract_packet { union abstract_packet {
@ -45,10 +43,7 @@ struct internet_packet {
}; };
struct buffer_packet { struct buffer_packet {
uint64_t ref_count;
uint8_t mode; uint8_t mode;
int fdread;
int fdwrite;
uint16_t aread; uint16_t aread;
uint16_t awrite; uint16_t awrite;
struct internet_packet ip; struct internet_packet ip;