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

Rework algorithm

Browse source
This commit is contained in:
Quentin Dufour 2019-03-19 15:39:05 +01:00
parent d89b7151c9
commit 5853e99e7b
3 changed files with 70 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

137
src/algo_rr.c
View file

@ -3,7 +3,12 @@
#include "algo_utils.h"
#include "utils.h"
#define WAITING 10
struct waited_pkt {
uint16_t id;
int link_num;
int timer_fd;
uint8_t on;
};
struct deferred_pkt {
int link_fd;
@ -18,8 +23,8 @@ struct rr_ctx {
uint16_t recv_id;
uint16_t sent_id;
uint8_t current_link;
struct deferred_pkt real[WAITING];
struct buffer_packet stub_bp;
struct deferred_pkt real[PACKET_BUFFER_SIZE];
struct waited_pkt wait[PACKET_BUFFER_SIZE];
};
int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
@ -64,7 +69,7 @@ co_error:
exit(EXIT_FAILURE);
}
void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, int64_t link) {
int set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, struct waited_pkt* wpkt) {
struct timespec now;
struct itimerspec timer_config;
char url[1024];
@ -93,13 +98,14 @@ void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, int64_t link) {
exit(EXIT_FAILURE);
}
fdinfo.cat->name = "timeout";
fdinfo.other = (void*)link; // Should put the link number and the id
fdinfo.other = wpkt; // Should put the link number and the id
fdinfo.free_other = NULL;
sprintf(fdinfo.url, "timer:%ld:1", micro_sec);
evt_core_add_fd (evts, &fdinfo);
return fdinfo.fd;
}
int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
void rr_pkt_register(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;
@ -109,57 +115,59 @@ int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct
rr->my_links_ver = bp->ip.ap.str.id;
}
// 2. If packet arrived too late
// 2. If packet arrived too late, we discard it
if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) {
if (bp->ip.ap.str.flags & PKT_TIMEOUT) return 0; // We don't use real buffer pkt for PKT_TIMEOUT
// 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;
mv_buffer_wtor (app_ctx, fdinfo, bp);
return;
}
// 3. If packet arrived too early
// 3. If packet arrived too early, we register a timer
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, bp->ip.ap.str.prevlink);
// 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 % PACKET_BUFFER_SIZE;
rr->real[idx].bp = bp;
rr->real[idx].link_fd = fdinfo->fd;
return 0;
int idx_waited = (bp->ip.ap.str.id - 1) % PACKET_BUFFER_SIZE;
rr->wait[idx_waited].on = 1;
rr->wait[idx_waited].id = bp->ip.ap.str.id;
rr->wait[idx_waited].link_num = bp->ip.ap.str.prevlink;
set_timeout(ctx, timeout, &rr->wait[idx_waited]);
}
// 4. If we were waiting this packet
rr->recv_id = bp->ip.ap.str.id;
// 4. We queue the packet
int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
rr->real[idx_real].bp = bp;
rr->real[idx_real].link_fd = fdinfo->fd;
// 5. We make sure that the remote link is set to up
char buffer[16];
url_get_port (buffer, fdinfo->url);
int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
// 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;
rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working
}
void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
void rr_deliver(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;
struct rr_ctx* rr = app_ctx->misc;
char url[255];
// 1. A whole packet has been read, we will find its target
// 0. We update our cursor
rr->recv_id = bp->ip.ap.str.id;
// 1. We check that we don't have a running timeout
int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
if (rr->wait[idx_real].on) {
rr->wait[idx_real].on = 0;
evt_core_rm_fd (ctx, rr->wait[idx_real].timer_fd);
}
// 2. We free the buffer if it's a control packet and quit
if (bp->ip.ap.str.flags & PKT_CONTROL) {
mv_buffer_wtor (app_ctx, fdinfo, bp);
return;
}
// 3. 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) {
@ -167,33 +175,36 @@ void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_c
mv_buffer_wtor (app_ctx, fdinfo, bp);
}
// 2. We move the buffer and notify the target
// 4. 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;
void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
struct rr_ctx* rr = app_ctx->misc;
struct evt_core_fdinfo* fdinfo = NULL;
struct buffer_packet* bp = NULL;
do {
if(rr_update_states(ctx, bp, fdinfo)) rr_deliver(ctx, bp, fdinfo);
do {
struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE];
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 fd\n");
rr->recv_id++;
}
} while (fdinfo == NULL);
} while(1);
while(1) {
struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE];
if (def->bp == NULL) break;
bp = def->bp;
def->bp = NULL;
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 fd\n");
rr->recv_id++;
continue;
}
rr_deliver(ctx, fdinfo, bp);
}
}
int rr_on_tcp_read(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;
int read_res = FDS_READY;
// 1. Get current read buffer OR a new read buffer OR subscribe to be notified later
@ -207,7 +218,8 @@ int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
}
// 3. Logic on packet
rr_pkt_recv (ctx, fdinfo, bp);
rr_pkt_register(ctx, fdinfo, bp);
rr_pkt_unroll (ctx, app_ctx);
return 0;
co_error:
@ -242,22 +254,9 @@ co_error:
int rr_on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
struct rr_ctx* rr = app_ctx->misc;
uint16_t id;
evt_core_rm_fd (ctx, fdinfo->fd); // We use the timeout only once
int fd = (int)fdinfo->other;
struct evt_core_fdinfo* tfdinfo = evt_core_get_from_fd (ctx, fd);
if (tfdinfo == NULL) {
fprintf(stderr, "An error occured as the link seems to be closed for the requested fd\n");
rr->recv_id++;
return 1;
}
rr->stub_bp.ip.ap.str.id = 0; //?
rr->stub_bp.ip.ap.str.flags = PKT_TIMEOUT;
rr->stub_bp.ip.ap.str.deltat = rr->mjit;
rr->stub_bp.ip.ap.str.prevlink = UINT8_MAX;
rr_pkt_recv (ctx, tfdinfo, &rr->stub_bp);
return 1;
}

2
src/algo_utils.h
View file

@ -3,7 +3,7 @@
#include <glib-2.0/glib.h>
#include <glib-2.0/gmodule.h>
#include <glib-2.0/glib-object.h>
#define PACKET_BUFFER_SIZE 10
#define PACKET_BUFFER_SIZE 100
typedef void (*algo_ctx_free_misc)(void*);

3
src/packet.h
View file

@ -29,8 +29,7 @@ enum BP_MODE {
};
enum PKT_FLAGS {
PKT_TIMEOUT = 1 << 0,
PKT_CONTROL = 1 << 1
PKT_CONTROL = 1 << 0
};
union abstract_packet {