tor_multipath_voip/src/algo_rr.c

342 lines
11 KiB
C
Raw Normal View History

2019-03-13 16:53:46 +00:00
#include <sys/timerfd.h>
2019-03-07 15:57:02 +00:00
#include "algo_utils.h"
2019-03-18 16:58:40 +00:00
#include "utils.h"
2019-04-24 14:23:41 +00:00
#include "url.h"
#include "proxy.h"
2019-05-07 17:18:27 +00:00
#include "timer.h"
2019-03-18 09:26:02 +00:00
2019-03-19 14:39:05 +00:00
struct waited_pkt {
uint16_t id;
int link_num;
uint8_t on;
2019-05-07 17:18:27 +00:00
struct algo_ctx* algo;
2019-03-19 14:39:05 +00:00
};
2019-03-19 12:50:38 +00:00
2019-03-18 09:26:02 +00:00
struct deferred_pkt {
int link_fd;
2019-03-20 14:13:16 +00:00
int idx;
2019-03-27 17:27:44 +00:00
uint16_t id;
2019-03-20 14:13:16 +00:00
uint8_t on;
2019-05-07 17:18:27 +00:00
struct algo_ctx* algo;
2019-03-18 09:26:02 +00:00
};
struct rr_ctx {
uint8_t my_links;
uint16_t my_links_ver;
uint8_t remote_links;
2019-03-19 17:50:56 +00:00
int64_t mjit;
2019-03-18 09:26:02 +00:00
uint16_t recv_id;
2019-03-27 15:24:39 +00:00
uint16_t recv_id_late;
2019-03-18 09:26:02 +00:00
uint16_t sent_id;
2019-03-18 16:58:40 +00:00
uint8_t current_link;
2019-05-09 17:53:47 +00:00
struct internet_packet prev_packet;
2019-03-19 15:35:11 +00:00
struct timespec emit_time;
2019-03-19 14:39:05 +00:00
struct deferred_pkt real[PACKET_BUFFER_SIZE];
struct waited_pkt wait[PACKET_BUFFER_SIZE];
2019-03-18 09:26:02 +00:00
};
2019-03-27 10:48:09 +00:00
void show_link_availability(struct rr_ctx* rr) {
printf("Links availability: my_links[");
for (int i = 0; i < 8; i++) {
if (rr->my_links & 1 << i) printf("U");
else printf("-");
}
printf("], rem_links[");
for (int i = 0; i < 8; i++) {
if (rr->remote_links & 1 << i) printf("U");
else printf("-");
}
printf("]\n");
}
2019-05-07 17:18:27 +00:00
void expired_wait (struct evt_core_ctx* ctx, void* user);
void expired_late(struct evt_core_ctx* ctx, void* user);
2019-03-13 16:53:46 +00:00
2019-03-19 14:39:05 +00:00
void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
2019-03-13 16:53:46 +00:00
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
2019-03-18 09:26:02 +00:00
struct rr_ctx* rr = app_ctx->misc;
2019-03-27 15:24:39 +00:00
char buffer[16];
url_get_port (buffer, fdinfo->url);
int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
2019-05-09 17:53:47 +00:00
uint16_t real_idx = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
uint16_t wait_idx = (bp->ip.ap.str.id - 1) % PACKET_BUFFER_SIZE;
2019-03-27 15:24:39 +00:00
2019-03-27 16:46:10 +00:00
//printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id);
2019-03-13 16:53:46 +00:00
2019-03-27 15:24:39 +00:00
// 0. Update remote links
if (ring_lt(rr->recv_id_late, bp->ip.ap.str.id) && !(rr->remote_links & 1 << link_num)) {
2019-03-28 14:29:34 +00:00
printf("Activate link=%d | ", link_num);
2019-03-27 15:24:39 +00:00
rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working
show_link_availability (rr);
}
2019-03-19 16:54:28 +00:00
2019-03-27 15:24:39 +00:00
// 1. Update my links I can use thanks to target feedback
2019-03-27 17:11:05 +00:00
if (bp->ip.ap.str.id > rr->my_links_ver && bp->ip.ap.str.bitfield != rr->my_links) {
2019-03-18 16:58:40 +00:00
rr->my_links = bp->ip.ap.str.bitfield;
rr->my_links_ver = bp->ip.ap.str.id;
2019-03-28 14:29:34 +00:00
printf("Update my links | ");
2019-03-27 17:11:05 +00:00
show_link_availability (rr);
2019-03-13 16:53:46 +00:00
}
2019-05-09 17:53:47 +00:00
// 2. If packet arrived too late or already queued, we discard it
if (ring_ge(rr->recv_id, bp->ip.ap.str.id) || rr->real[real_idx].id == bp->ip.ap.str.id) {
2019-03-13 16:53:46 +00:00
// Packet has already been delivered or dropped, we free the buffer
2019-05-09 17:53:47 +00:00
fprintf(stderr, "Packet %d arrived too late (current: %d) or already received\n", bp->ip.ap.str.id, rr->recv_id);
2019-03-20 14:13:16 +00:00
mv_buffer_wtof (app_ctx, fdinfo);
2019-03-19 14:39:05 +00:00
return;
2019-03-13 16:53:46 +00:00
}
2019-05-07 17:18:27 +00:00
// 3. If packet arrived too early, we wait for its predecessors
2019-03-20 09:28:52 +00:00
//printf("%d < %d = %d\n", rr->recv_id, bp->ip.ap.str.id - 1, ring_lt(rr->recv_id, bp->ip.ap.str.id - 1));
2019-03-18 16:58:40 +00:00
if (ring_lt(rr->recv_id, bp->ip.ap.str.id - 1)) {
2019-03-18 09:26:02 +00:00
int64_t timeout = rr->mjit - (int64_t) bp->ip.ap.str.deltat;
2019-03-19 19:48:07 +00:00
//printf("%ld - %ld = %ld\n", rr->mjit, (int64_t) bp->ip.ap.str.deltat, timeout);
2019-03-13 16:53:46 +00:00
if (timeout <= 0) timeout = 0;
2019-03-15 15:44:47 +00:00
2019-05-07 17:18:27 +00:00
if (rr->wait[wait_idx].on && rr->wait[wait_idx].id != bp->ip.ap.str.id - 1) {
fprintf(stderr, "Waiting array overlap, BUG: [\n");
for (int i = 0; i < PACKET_BUFFER_SIZE; i++) {
printf("\t%d => %d\n", rr->wait[i].id, rr->wait[i].on);
}
printf("] - could be replaced by drop\n");
exit(EXIT_FAILURE);
} else if (!rr->wait[wait_idx].on) {
rr->wait[wait_idx].on = 1;
rr->wait[wait_idx].id = bp->ip.ap.str.id - 1;
rr->wait[wait_idx].link_num = bp->ip.ap.str.prevlink;
rr->wait[wait_idx].algo = app_ctx;
set_timeout(ctx, timeout, &rr->wait[wait_idx], expired_wait);
}
2019-03-27 15:24:39 +00:00
}
2019-05-07 17:18:27 +00:00
// 4. We queue the packet to keep it
if (rr->real[real_idx].on && rr->real[real_idx].id != bp->ip.ap.str.id) {
fprintf(stderr, "Real array is full for id=%d, idx=%d, BUG: [\n", bp->ip.ap.str.id, real_idx);
for (int i = 0; i < PACKET_BUFFER_SIZE; i++) {
2019-05-07 17:18:27 +00:00
printf("\t%d => %d\n", rr->real[i].id, rr->real[i].on);
}
2019-05-07 17:18:27 +00:00
printf("] - could be replaced by drop\n");
2019-03-27 17:27:44 +00:00
exit(EXIT_FAILURE);
2019-05-07 17:18:27 +00:00
} else if (!rr->real[real_idx].on) {
rr->real[real_idx].on = 2;
rr->real[real_idx].id = bp->ip.ap.str.id;
rr->real[real_idx].idx = real_idx;
rr->real[real_idx].link_fd = fdinfo->fd;
rr->real[real_idx].algo = app_ctx;
mv_buffer_rtoa(app_ctx, fdinfo, &rr->real[real_idx].idx);
// 5. We register a timer for this packet to create a reactivation window for broken links
set_timeout(ctx, rr->mjit + 1, &rr->real[real_idx], expired_late);
//printf("%d is added to real as %d\n", bp->ip.ap.str.id, idx_real);
2019-03-27 17:27:44 +00:00
} else {
fprintf(stdout, "Packet %d already received (current: %d)\n", bp->ip.ap.str.id, rr->recv_id);
mv_buffer_wtof (app_ctx, fdinfo);
}
2019-05-07 17:18:27 +00:00
2019-03-18 16:58:40 +00:00
}
2019-03-20 14:13:16 +00:00
void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct deferred_pkt* dp) {
2019-03-18 16:58:40 +00:00
struct evt_core_fdinfo *to_fdinfo = NULL;
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
2019-03-19 14:39:05 +00:00
struct rr_ctx* rr = app_ctx->misc;
2019-03-18 16:58:40 +00:00
char url[255];
2019-03-20 14:13:16 +00:00
// 1. Marked the packet as handled
2019-05-07 17:18:27 +00:00
dp->on--;
2019-03-20 14:13:16 +00:00
2019-05-07 17:18:27 +00:00
// 2. Get the buffer
2019-03-20 14:13:16 +00:00
struct buffer_packet* bp = get_app_buffer (app_ctx, &dp->idx);
2019-03-19 16:54:28 +00:00
2019-03-20 14:13:16 +00:00
// 3. We update our cursor
2019-03-19 14:39:05 +00:00
rr->recv_id = bp->ip.ap.str.id;
2019-05-07 17:18:27 +00:00
// 4. Find its target
2019-03-18 16:58:40 +00:00
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);
2019-03-20 14:13:16 +00:00
//mv_buffer_wtor (app_ctx, fdinfo, bp);
mv_buffer_atof (app_ctx, &dp->idx);
2019-03-18 16:58:40 +00:00
}
2019-05-07 17:18:27 +00:00
// 5. We move the buffer and notify the target
2019-03-20 14:13:16 +00:00
//mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
mv_buffer_atow (app_ctx, &dp->idx, to_fdinfo);
2019-04-24 14:23:41 +00:00
main_on_udp_write(ctx, to_fdinfo);
2019-03-18 16:58:40 +00:00
}
2019-03-19 14:39:05 +00:00
void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
2019-03-18 16:58:40 +00:00
struct rr_ctx* rr = app_ctx->misc;
2019-03-19 14:39:05 +00:00
struct evt_core_fdinfo* fdinfo = NULL;
struct buffer_packet* bp = NULL;
while(1) {
2019-03-27 16:46:10 +00:00
//printf("Trying to deliver %d\n", rr->recv_id+1);
2019-03-19 14:39:05 +00:00
struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE];
2019-03-20 14:13:16 +00:00
if (!def->on) break;
2019-03-19 14:39:05 +00:00
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;
}
2019-03-20 14:13:16 +00:00
rr_deliver(ctx, fdinfo, def);
2019-03-27 16:46:10 +00:00
//printf("Delivered %d\n", rr->recv_id);
2019-03-19 14:39:05 +00:00
}
2019-03-13 16:53:46 +00:00
}
2019-03-19 15:35:11 +00:00
//------
2019-04-24 14:23:41 +00:00
int algo_rr_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
2019-03-13 16:53:46 +00:00
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
2019-03-19 14:39:05 +00:00
struct rr_ctx* rr = app_ctx->misc;
2019-03-13 16:53:46 +00:00
2019-04-24 14:23:41 +00:00
// 1. Register packet in our queue
2019-03-19 14:39:05 +00:00
rr_pkt_register(ctx, fdinfo, bp);
2019-04-24 14:23:41 +00:00
// 2. Process queue
2019-03-19 14:39:05 +00:00
rr_pkt_unroll (ctx, app_ctx);
2019-03-13 16:53:46 +00:00
return 0;
co_error:
perror("Failed to TCP read");
exit(EXIT_FAILURE);
}
2019-04-24 14:23:41 +00:00
int algo_rr_on_datagram(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;
struct evt_core_fdinfo *to_fdinfo = NULL;
char url[255];
2019-03-19 15:35:11 +00:00
2019-04-24 14:23:41 +00:00
// 1. Prepare RR state and packet values
2019-03-19 15:35:11 +00:00
struct timespec curr;
int secs, nsecs;
2019-03-19 17:50:56 +00:00
uint64_t mili_sec;
2019-03-19 15:35:11 +00:00
if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
perror("clock_gettime error");
exit(EXIT_FAILURE);
}
secs = curr.tv_sec - rr->emit_time.tv_sec;
nsecs = curr.tv_nsec - rr->emit_time.tv_nsec;
2019-03-19 17:50:56 +00:00
mili_sec = secs * 1000 + nsecs / 1000000;
if (mili_sec > rr->mjit) mili_sec = rr->mjit;
2019-03-19 15:35:11 +00:00
bp->ip.ap.str.id = rr->sent_id;
bp->ip.ap.str.flags = 0;
2019-03-19 17:50:56 +00:00
bp->ip.ap.str.deltat = mili_sec;
2019-03-19 15:35:11 +00:00
bp->ip.ap.str.bitfield = rr->remote_links;
bp->ip.ap.str.prevlink = rr->current_link;
2019-05-09 17:53:47 +00:00
if (app_ctx->ap.redundant_data == 1) {
append_buffer(&bp->ip.ap, 1, &rr->prev_packet.ap); // We append previous packet
append_buffer(&rr->prev_packet.ap, 0, &bp->ip.ap); // We store current packet for next time
bp->ap_count++;
}
2019-03-27 16:46:10 +00:00
//printf("Will send packet id=%d\n", bp->ip.ap.str.id);
2019-03-19 15:35:11 +00:00
rr->emit_time = curr;
rr->sent_id++;
2019-04-24 14:23:41 +00:00
// 2. Try to find someone to send it
2019-05-09 17:53:47 +00:00
int max = 16;
2019-03-27 15:24:39 +00:00
uint8_t sel_link = rr->current_link;
while(max-- >= 0) {
2019-05-09 13:33:27 +00:00
sel_link = (sel_link + 1) % app_ctx->ap.links;
2019-03-27 15:24:39 +00:00
sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + sel_link); //@FIXME Hardcoded
to_fdinfo = evt_core_get_from_url (ctx, url);
if (to_fdinfo == NULL) continue; // Missing link
if (app_ctx->ap.is_waiting_bootstrap && !app_ctx->is_rdy) goto not_ready; // Some links are down
if (!app_ctx->ap.is_healing || rr->my_links & (1 << sel_link)) {
2019-03-27 15:24:39 +00:00
rr->current_link = sel_link;
mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
2019-04-24 14:23:41 +00:00
main_on_tcp_write(ctx, to_fdinfo);
2019-03-27 15:24:39 +00:00
return 0;
} else {
dup_buffer_tow(app_ctx, bp, to_fdinfo);
2019-04-24 14:23:41 +00:00
main_on_tcp_write(ctx, to_fdinfo);
2019-03-27 15:24:39 +00:00
}
2019-03-19 15:35:11 +00:00
}
2019-03-28 14:26:33 +00:00
not_ready:
2019-04-24 14:23:41 +00:00
// 3. We find no up target
2019-03-28 14:26:33 +00:00
fprintf(stderr, "Still bootstrapping or no link to forward data from %s in udp-read. Dropping packet :( \n", fdinfo->url);
2019-03-27 15:24:39 +00:00
mv_buffer_wtof (app_ctx, fdinfo);
2019-03-19 15:35:11 +00:00
return 0;
co_error:
perror("Failed to UDP read");
exit(EXIT_FAILURE);
}
2019-05-07 17:18:27 +00:00
void expired_wait(struct evt_core_ctx* ctx, void* user) {
struct waited_pkt* pkt = user;
struct rr_ctx* rr = pkt->algo->misc;
2019-03-19 12:50:38 +00:00
2019-05-07 17:18:27 +00:00
// 1. Release lock
2019-03-19 20:21:00 +00:00
pkt->on = 0;
2019-05-07 17:18:27 +00:00
// 2. We will not reactivate link for this packet
if (ring_lt(rr->recv_id_late, pkt->id)) rr->recv_id_late = pkt->id;
2019-05-13 08:02:43 +00:00
/*
2019-05-07 17:18:27 +00:00
// 3. Stop if packet has been received and delivered
if (ring_le (pkt->id, rr->recv_id)) return;
2019-03-19 20:06:01 +00:00
printf("Timer reached for packet %d\n", pkt->id);
2019-05-13 08:02:43 +00:00
*/
2019-03-19 12:50:38 +00:00
2019-05-07 17:18:27 +00:00
// 4. BLACKLIST LINK
2019-03-28 14:29:34 +00:00
printf("Blacklist link=%d | ", pkt->link_num);
2019-03-27 10:19:05 +00:00
rr->remote_links &= 0xff ^ 1 << pkt->link_num;
2019-03-27 13:17:41 +00:00
show_link_availability (rr);
2019-03-20 09:28:52 +00:00
2019-05-07 17:18:27 +00:00
// 5. Deliver following packets
2019-03-20 08:08:32 +00:00
while (ring_lt(rr->recv_id, pkt->id)) {
rr->recv_id++;
2019-05-07 17:18:27 +00:00
rr_pkt_unroll (ctx, pkt->algo);
2019-03-20 08:08:32 +00:00
}
2019-05-07 17:18:27 +00:00
}
void expired_late(struct evt_core_ctx* ctx, void* user) {
struct deferred_pkt* pkt = user;
struct rr_ctx* rr = pkt->algo->misc;
2019-03-19 14:39:05 +00:00
2019-05-07 17:18:27 +00:00
pkt->on--;
if (ring_lt(rr->recv_id_late, pkt->id)) rr->recv_id_late = pkt->id;
2019-03-19 12:50:38 +00:00
}
2019-04-24 14:23:41 +00:00
int algo_rr_on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
// We do nothing
2019-03-19 09:00:03 +00:00
return 0;
}
2019-05-09 09:39:03 +00:00
void algo_rr_free(void* v) {
struct rr_ctx* rr = v;
free(rr);
}
2019-04-24 14:23:41 +00:00
void algo_rr_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, struct algo_params* ap) {
2019-03-18 09:26:02 +00:00
struct rr_ctx* rr = malloc(sizeof(struct rr_ctx));
2019-04-24 14:23:41 +00:00
if (rr == NULL) {
perror("malloc failed for rr_init.");
exit(EXIT_FAILURE);
}
2019-03-18 09:26:02 +00:00
memset(rr, 0, sizeof(struct rr_ctx));
rr->mjit = 200;
rr->my_links = 0xff;
rr->remote_links = 0xff;
2019-03-19 16:54:28 +00:00
rr->sent_id = 1;
2019-03-27 15:24:39 +00:00
rr->recv_id = 0;
rr->recv_id_late = 0;
2019-04-24 14:23:41 +00:00
app_ctx->misc = rr;
2019-05-09 09:39:03 +00:00
app_ctx->free_misc = algo_rr_free;
2019-05-07 17:18:27 +00:00
init_timer(ctx);
2019-03-07 15:57:02 +00:00
}
2019-04-24 14:23:41 +00:00