tor_multipath_voip/src/algo_utils.c

225 lines
7 KiB
C
Raw Normal View History

2019-03-07 15:57:02 +00:00
#include "algo_utils.h"
void free_nothing(void* app_ctx) {}
void free_naive(void* app_ctx) {
struct algo_ctx* ctx = (struct algo_ctx*) app_ctx;
ctx->ref_count--;
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);
g_hash_table_destroy (ctx->write_waiting);
free(ctx);
}
2019-03-20 09:28:52 +00:00
void iterate(int* fd, GQueue* q, int* waiting_count) {
fprintf(stderr, "Queue for fd=%d has length=%d\n", *fd, q->length);
waiting_count += q->length;
}
void iterate2(int* fd, struct buffer_packet *bp, gpointer user_data) {
fprintf(stderr, "fd=%d has a used_buffer entry\n", *fd);
}
2019-03-27 15:24:39 +00:00
void debug_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
fprintf(stderr, "No more free buffer for fd=%d.\n", fdinfo->fd);
int waiting_count = 0;
g_hash_table_foreach(app_ctx->write_waiting, (GHFunc)iterate, &waiting_count);
g_hash_table_foreach(app_ctx->used_buffer, (GHFunc)iterate2, NULL);
fprintf(stderr, "total_buffers=%d, free_buffer=%d, used_buffers=%d, app_buffer=%d, write_buffer=%d.\n",
PACKET_BUFFER_SIZE,
app_ctx->free_buffer->length,
g_hash_table_size(app_ctx->used_buffer),
g_hash_table_size(app_ctx->application_waiting),
waiting_count);
}
2019-03-07 15:57:02 +00:00
/**
* Returns a buffer if available, NULL otherwise
*/
struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
struct buffer_packet* bp;
// 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. Get a new buffer otherwise
bp = g_queue_pop_head(app_ctx->free_buffer);
if (bp == NULL) {
2019-03-27 15:24:39 +00:00
debug_buffer(app_ctx, fdinfo);
2019-03-07 15:57:02 +00:00
// 2.1 If no buffer is available, we subscribe to be notified later
g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd));
return NULL;
}
// 3. Update state
g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp);
return bp;
}
/**
* Returns a buffer if available, NULL otherwise
*/
struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
struct buffer_packet* bp;
GQueue* q;
// 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;
}
2019-03-20 14:13:16 +00:00
void mv_buffer_rtow(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, struct evt_core_fdinfo* to) {
2019-03-07 15:57:02 +00:00
GQueue* q;
2019-03-20 14:13:16 +00:00
struct buffer_packet* bp;
// 1. We get the packet buffer
bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd);
if (bp == NULL) {
fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", from->fd, from->url);
exit(EXIT_FAILURE);
}
// 2. We get the target writing queue
2019-03-07 15:57:02 +00:00
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);
}
2019-03-20 14:13:16 +00:00
// 3. We move the data
2019-03-07 15:57:02 +00:00
g_hash_table_remove(app_ctx->used_buffer, &from->fd);
g_queue_push_tail(q, bp);
}
2019-03-20 14:13:16 +00:00
void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd));
if (bp == NULL) {
fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", fdinfo->fd, fdinfo->url);
exit(EXIT_FAILURE);
}
2019-03-07 15:57:02 +00:00
bp->mode = BP_READING;
bp->aread = 0;
g_queue_push_tail (app_ctx->free_buffer, bp);
g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd));
}
2019-03-20 14:13:16 +00:00
void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to) {
struct buffer_packet* bp;
bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd);
if (bp == NULL) {
fprintf(stderr, "Unable to find a buffer for fd=%d url=%s\n", from->fd, from->url);
exit(EXIT_FAILURE);
}
g_hash_table_remove(app_ctx->used_buffer, &from->fd);
if (g_hash_table_contains(app_ctx->application_waiting, to)) {
2019-03-27 16:31:46 +00:00
fprintf(stderr, "Data already exists for this entry\n");
debug_buffer(app_ctx, from);
2019-03-20 14:13:16 +00:00
exit(EXIT_FAILURE);
}
g_hash_table_insert(app_ctx->application_waiting, to, bp);
}
void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to) {
GQueue* q;
struct buffer_packet* bp;
// 1. We get the buffer
bp = g_hash_table_lookup (app_ctx->application_waiting, from);
if (bp == NULL) {
fprintf(stderr, "Unable to find this application buffer\n");
exit(EXIT_FAILURE);
}
// 2. We get the target writing queue
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);
}
// 3. We move the buffer
g_hash_table_remove (app_ctx->application_waiting, from);
g_queue_push_tail(q, bp);
}
void mv_buffer_atof(struct algo_ctx* app_ctx, void* from) {
struct buffer_packet* bp;
// 1. We get the buffer
bp = g_hash_table_lookup (app_ctx->application_waiting, from);
if (bp == NULL) {
fprintf(stderr, "Unable to find this application buffer\n");
exit(EXIT_FAILURE);
}
// 2. We move it
g_hash_table_remove (app_ctx->application_waiting, from);
g_queue_push_tail (app_ctx->free_buffer, bp);
}
2019-03-27 15:24:39 +00:00
void dup_buffer_tow(struct algo_ctx* app_ctx, struct buffer_packet* bp, struct evt_core_fdinfo* to) {
GQueue* q;
// 1. We get a free buffer
struct buffer_packet* bp_dest = g_queue_pop_head(app_ctx->free_buffer);
if (bp_dest == NULL) {
debug_buffer(app_ctx, to);
return;
}
// 2. We duplicate the data
memcpy(bp_dest, bp, sizeof(struct buffer_packet));
// 3. We get the target writing queue
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);
}
// 4. We push the content to the appropriate destination
g_queue_push_tail(q, bp_dest);
}
2019-03-20 14:13:16 +00:00
struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx) {
return g_hash_table_lookup (app_ctx->application_waiting, idx);
}
2019-03-07 15:57:02 +00:00
void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
struct evt_core_fdinfo* next_fdinfo = NULL;
while (next_fdinfo == NULL) {
2019-03-20 09:28:52 +00:00
int* fd = g_queue_pop_head(app_ctx->read_waiting);
if (fd == NULL) break;
next_fdinfo = evt_core_get_from_fd (ctx, *fd);
if (next_fdinfo == NULL) {
fprintf(stderr, "Unable to find fdinfo for fd=%d\n", *fd);
exit(EXIT_FAILURE);
} else if (strcmp(next_fdinfo->cat->name, "tcp-read") == 0 || strcmp(next_fdinfo->cat->name, "udp-read") == 0) {
2019-03-07 15:57:02 +00:00
next_fdinfo->cat->cb(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 naive_free_simple(void* v) {
2019-04-04 08:44:20 +00:00
GQueue* g = v;
g_queue_free (g);
2019-03-07 15:57:02 +00:00
}