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

WIP algo rr, TCP read logic

Browse source
This commit is contained in:
Quentin Dufour 2019-03-18 17:58:40 +01:00
parent 02a3507ef5
commit 588219f0f0
3 changed files with 118 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

124
src/algo_rr.c
View file

@ -1,7 +1,7 @@
#include <sys/timerfd.h>
#include "algo_skel.h"
#include "algo_utils.h"
#include "utils.h"
struct deferred_pkt {
int link_fd;
@ -15,10 +15,15 @@ struct rr_ctx {
uint16_t mjit;
uint16_t recv_id;
uint16_t sent_id;
uint8_t current_link;
struct deferred_pkt real[LINK_COUNT];
struct deferred_pkt stub[LINK_COUNT];
};
int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
int rr_on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
int rr_on_udp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
int rr_on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
int conn_sock1, conn_sock2;
@ -92,54 +97,99 @@ void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec) {
evt_core_add_fd (evts, &fdinfo);
}
void rr_handle_recv(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
struct rr_ctx* rr = app_ctx->misc;
char buffer[16];
url_get_port (buffer, fdinfo->url);
int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
// 1. Update links I can use thanks to target feedback
if (bp->ip.ap.str.bitfield ^ rr->my_links) {
update_my_bitfield(bp, rr);
if (bp->ip.ap.str.id > rr->my_links_ver) {
rr->my_links = bp->ip.ap.str.bitfield;
rr->my_links_ver = bp->ip.ap.str.id;
}
// 2. If packet arrived too late
if (rr->recv_id > bp->ip.ap.str.id - 1) {
if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) {
// Packet has already been delivered or dropped, we free the buffer
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);
return 0;
}
// 3. If packet arrived too early
else if (rr->recv_id < bp->ip.ap.str.id - 1) {
if (ring_lt(rr->recv_id, bp->ip.ap.str.id - 1)) {
int64_t timeout = rr->mjit - (int64_t) bp->ip.ap.str.deltat;
if (timeout <= 0) timeout = 0;
set_timeout(ctx, timeout);
// Add a buffer to stub too
// Bitfield can be anything as a greater packet has triggered the bitfield update before
int idx = bp->ip.ap.str.id % LINK_COUNT;
rr->real[idx].bp = bp;
rr->real[idx].link_fd = fdinfo->fd;
return 0;
}
// 4. If we were waiting this packet
else {
if (bp->ip.ap.str.flags & PKT_TIMEOUT) broken_rlink(rr, fdinfo);
else if (bp->ip.ap.str.flags & PKT_CONTROL) working_rlink(rr, fdinfo);
else deliver(bp);
rr->recv_id = bp->ip.ap.str.id;
struct deferred_pkt* def = &rr->real[rr->recv_id+1];
if (def->bp != NULL) {
struct evt_core_fdinfo* next = evt_core_get_from_fd (ctx, def->link_fd);
if (next == NULL) return;
rr_handle_recv(ctx, def->bp, next);
// 4.1 This is a timeout packet, we set the link as dead
if (bp->ip.ap.str.flags & PKT_TIMEOUT) {
rr->remote_links &= (1 << link_num) ^ UINT16_MAX;
return 0;
}
// 4.2 This is a control packet, we set the link as alive
if (bp->ip.ap.str.flags & PKT_CONTROL) {
rr->remote_links |= 1 << link_num;
return 0;
}
return 1;
}
void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
struct evt_core_fdinfo *to_fdinfo = NULL;
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
char url[255];
// 1. A whole packet has been read, we will find its target
sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port);
to_fdinfo = evt_core_get_from_url (ctx, url);
if (to_fdinfo == NULL) {
fprintf(stderr, "No fd for URL %s in udp:write for tcp-read. Dropping packet :( \n", url);
mv_buffer_wtor (app_ctx, fdinfo, bp);
}
// 2. We move the buffer and notify the target
mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
rr_on_udp_write(ctx, to_fdinfo);
}
void rr_pkt_recv(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
struct rr_ctx* rr = app_ctx->misc;
do {
if(rr_update_states(ctx, bp, fdinfo)) rr_deliver(ctx, bp, fdinfo);
do {
struct deferred_pkt* def = &rr->real[rr->recv_id+1 % LINK_COUNT];
if (def->bp == NULL) break;
def->bp = NULL;
struct evt_core_fdinfo* fdinfo = evt_core_get_from_fd (ctx, def->link_fd);
if (fdinfo == NULL) {
fprintf(stderr, "An error occured as the link seems to be closed for the requested link\n");
rr->recv_id++;
}
} while (fdinfo == NULL);
} while(1);
}
int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp;
struct evt_core_fdinfo *to_fdinfo = NULL;
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
int read_res = FDS_READY;
char url[255];
// 1. Get current read buffer OR a new read buffer OR subscribe to be notified later
if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return 1;
@ -152,29 +202,37 @@ int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
}
// 3. Logic on packet
rr_handle_recv(ctx, bp, fdinfo);
rr_pkt_recv (ctx, fdinfo, bp);
/*
// 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);
to_fdinfo = evt_core_get_from_url (ctx, url);
if (to_fdinfo == NULL) {
fprintf(stderr, "No fd for URL %s in tcp-read. Dropping packet :( \n", url);
mv_buffer_wtor (app_ctx, fdinfo, bp);
return 1;
}
//printf("Pass packet from %s to %s\n", fdinfo->url, url);
// 4. We move the buffer and notify the target
mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
on_udp_write(ctx, to_fdinfo);
*/
return 0;
co_error:
perror("Failed to TCP read");
exit(EXIT_FAILURE);
}
int rr_on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp;
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
int write_res = FDS_READY;
// 1. Get current write buffer OR a buffer from the waiting queue OR leave
if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return 1;
// 2. Write buffer
write_res = write_packet_to_udp(fdinfo->fd, bp, fdinfo->other);
if (write_res == FDS_ERR) goto co_error;
if (write_res == FDS_AGAIN) return 1;
// 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 0;
co_error:
perror("Failed to UDP write");
exit(EXIT_FAILURE);
}
void algo_rr(struct algo_skel* as) {
struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx));

20
src/utils.c
View file

@ -60,3 +60,23 @@ int ring_buffer_free_space(struct ring_buffer* rb) {
int ring_buffer_used_space(struct ring_buffer* rb) {
return RING_BUFFER_SIZE - ring_buffer_free_space (rb);
}
int ring_gt(uint16_t v1, uint16_t v2) {
int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2;
return v1 != v2 && (vv1 - vv2) % UINT16_MAX <= UINT16_MAX / 2;
}
int ring_ge(uint16_t v1, uint16_t v2) {
int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2;
return (vv1 - vv2) % UINT16_MAX <= UINT16_MAX / 2;
}
int ring_lt(uint16_t v1, uint16_t v2) {
int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2;
return v1 != v2 && (vv1 - vv2) % UINT16_MAX > UINT16_MAX / 2;
}
int ring_le(uint16_t v1, uint16_t v2) {
int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2;
return (vv1 - vv2) % UINT16_MAX > UINT16_MAX / 2;
}

5
src/utils.h
View file

@ -18,3 +18,8 @@ void ring_buffer_ack_read(struct ring_buffer* rb, int size);
int ring_buffer_write(struct ring_buffer* rb, char* source, int size);
int ring_buffer_free_space(struct ring_buffer* rb);
int ring_buffer_used_space(struct ring_buffer* rb);
int ring_gt(uint16_t v1, uint16_t v2);
int ring_ge(uint16_t v1, uint16_t v2);
int ring_lt(uint16_t v1, uint16_t v2);
int ring_le(uint16_t v1, uint16_t v2);