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

All the pieces should be here

Browse source
This commit is contained in:
Quentin Dufour 2019-03-19 16:35:11 +01:00
parent 298e385dae
commit d48c5753fb
1 changed files with 99 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

105
src/algo_rr.c
View file

@ -23,6 +23,7 @@ struct rr_ctx {
uint16_t recv_id;
uint16_t sent_id;
uint8_t current_link;
struct timespec emit_time;
struct deferred_pkt real[PACKET_BUFFER_SIZE];
struct waited_pkt wait[PACKET_BUFFER_SIZE];
};
@ -202,6 +203,8 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
}
}
//------
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;
@ -252,6 +255,97 @@ co_error:
exit(EXIT_FAILURE);
}
int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
struct buffer_packet* bp;
struct evt_core_fdinfo *to_fdinfo;
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
struct rr_ctx* rr = app_ctx->misc;
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;
// 2. Read packet from socket
bp->ip.ap.str.port = url_get_port_int (fdinfo->url);
read_res = read_packet_from_udp (fdinfo->fd, bp, fdinfo->other);
if (read_res == FDS_ERR) goto co_error;
if (read_res == FDS_AGAIN) return 1;
// 3. Prepare RR state and packet values
struct timespec curr;
int secs, nsecs;
uint64_t micro_sec;
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;
micro_sec = secs * 1000000 + nsecs / 1000;
if (micro_sec > rr->mjit) micro_sec = rr->mjit;
bp->ip.ap.str.id = rr->sent_id;
bp->ip.ap.str.flags = 0;
bp->ip.ap.str.deltat = micro_sec;
bp->ip.ap.str.bitfield = rr->remote_links;
bp->ip.ap.str.prevlink = rr->current_link;
do {
rr->current_link = (rr->current_link) + 1 % 10;
} while (!(rr->my_links & (1 << rr->current_link)));
rr->emit_time = curr;
rr->sent_id++;
// 4. A whole packet has been read, we will find someone to write it
sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + rr->current_link); //@FIXME Hardcoded
to_fdinfo = evt_core_get_from_url (ctx, url);
if (to_fdinfo == NULL) {
fprintf(stderr, "No fd for URL %s in udp-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);
// 5. We move the buffer and notify the target
mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
rr_on_tcp_write(ctx, to_fdinfo);
return 0;
co_error:
perror("Failed to UDP read");
exit(EXIT_FAILURE);
}
int rr_on_tcp_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 data from the buffer to the socket
while (bp->mode == BP_WRITING) {
write_res = write_packet_to_tcp(fdinfo->fd, bp);
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 TCP write");
exit(EXIT_FAILURE);
}
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;
@ -327,23 +421,22 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) {
as->on_tcp_read.err_cb = rr_on_err;
ctx->ref_count++;
/*
as->on_udp_read.name = "udp-read";
as->on_udp_read.flags = EPOLLIN | EPOLLET;
as->on_udp_read.app_ctx = ctx;
as->on_udp_read.free_app_ctx = free_naive;
as->on_udp_read.cb = on_udp_read;
as->on_udp_read.err_cb = on_err;
as->on_udp_read.cb = rr_on_udp_read;
as->on_udp_read.err_cb = rr_on_err;
ctx->ref_count++;
as->on_tcp_write.name = "tcp-write";
as->on_tcp_write.flags = EPOLLOUT | EPOLLET | EPOLLRDHUP;
as->on_tcp_write.app_ctx = ctx;
as->on_tcp_write.free_app_ctx = free_naive;
as->on_tcp_write.cb = on_tcp_write;
as->on_tcp_write.err_cb = on_err;
as->on_tcp_write.cb = rr_on_tcp_write;
as->on_tcp_write.err_cb = rr_on_err;
ctx->ref_count++;
*/
as->on_udp_write.name = "udp-write";
as->on_udp_write.flags = EPOLLOUT | EPOLLET;