269 lines
8.5 KiB
C
269 lines
8.5 KiB
C
#include <sys/timerfd.h>
|
|
#include "algo_utils.h"
|
|
#include "utils.h"
|
|
#include "url.h"
|
|
#include "proxy.h"
|
|
#include "timer.h"
|
|
#include "proxy.h"
|
|
#define HISTORIC_SIZE 256
|
|
#define MAX_LINKS 64
|
|
|
|
enum ooo_state {
|
|
IN_ORDER,
|
|
OOO_ONGOING,
|
|
OOO_DONE
|
|
};
|
|
|
|
struct timing_entry {
|
|
enum ooo_state state;
|
|
struct timespec detected_at;
|
|
struct timespec finished_at;
|
|
uint8_t link_id;
|
|
uint64_t pkt_id;
|
|
};
|
|
|
|
struct light_ctx {
|
|
uint8_t prev_links[MAX_LINKS];
|
|
struct timing_entry historic[HISTORIC_SIZE];
|
|
struct timespec last[MAX_LINKS];
|
|
uint64_t pkt_rcv_id;
|
|
uint64_t pkt_sent_id;
|
|
uint8_t selected_link;
|
|
uint8_t total_links;
|
|
int max_ooo;
|
|
int sleep_duration;
|
|
int sent_past_links;
|
|
struct timespec window;
|
|
size_t monit_pkt_size;
|
|
};
|
|
|
|
void algo_lightning_free(void* v) {
|
|
struct light_ctx* lightc = v;
|
|
free(lightc);
|
|
}
|
|
|
|
void algo_lightning_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, struct algo_params* ap) {
|
|
app_ctx->misc = malloc(sizeof(struct light_ctx));
|
|
app_ctx->free_misc = algo_lightning_free;
|
|
if (app_ctx->misc == NULL) {
|
|
perror("malloc failed in algo lightning init");
|
|
exit(EXIT_FAILURE);
|
|
}
|
|
memset(app_ctx->misc, 0, sizeof(struct light_ctx));
|
|
struct light_ctx* lightc = app_ctx->misc;
|
|
lightc->total_links = app_ctx->ap.links;
|
|
lightc->selected_link = lightc->total_links - 1;
|
|
lightc->sent_past_links = lightc->total_links;
|
|
lightc->max_ooo = 50;
|
|
lightc->sleep_duration = 500;
|
|
|
|
uint64_t window = 5000;
|
|
if (ap->algo_specific_params != NULL) {
|
|
char *parse_ptr, *token, *params;
|
|
|
|
for (params = ap->algo_specific_params; ; params = NULL) {
|
|
token = strtok_r(params, ",", &parse_ptr);
|
|
if (token == NULL) break;
|
|
sscanf(token, "max_ooo=%d", &lightc->max_ooo);
|
|
sscanf(token, "recovery=%d", &lightc->sleep_duration);
|
|
sscanf(token, "window=%ld", &window);
|
|
sscanf(token, "sent_past_links=%d", &lightc->sent_past_links);
|
|
}
|
|
}
|
|
|
|
union abstract_packet m;
|
|
lightc->monit_pkt_size = sizeof(m.fmt.headers) + sizeof(m.fmt.content.link_monitoring_lightning) + sizeof(uint8_t) * (lightc->sent_past_links - 1);
|
|
timespec_set_unit (&lightc->window, window, MILISEC);
|
|
|
|
printf("max_ooo = %d ms\n", lightc->max_ooo);
|
|
printf("recovery = %d ms\n", lightc->sleep_duration);
|
|
printf("window check = %ld ms\n", window);
|
|
printf("sent_past_links = %d\n", lightc->sent_past_links);
|
|
}
|
|
|
|
void monitoring(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
|
|
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
|
|
struct light_ctx* lightc = app_ctx->misc;
|
|
union abstract_packet* ap = (union abstract_packet*) &bp->ip;
|
|
|
|
while (ap != NULL && ap->fmt.headers.cmd != CMD_LINK_MONITORING_LIGHTNING) ap = ap_next(ap);
|
|
if (ap == NULL) {
|
|
fprintf(stderr, "[algo_lightning] Unable to find our monitoring information\n");
|
|
exit(EXIT_FAILURE);
|
|
}
|
|
|
|
uint8_t *prev_links = &ap->fmt.content.link_monitoring_lightning.prev_links;
|
|
int64_t pkt_id = ap->fmt.content.link_monitoring_lightning.id;
|
|
int64_t missing = pkt_id - (lightc->pkt_rcv_id + 1);
|
|
struct timespec now;
|
|
set_now(&now);
|
|
|
|
// Detect OoO
|
|
for (int i = 0; i < missing && i < lightc->sent_past_links; i++) {
|
|
uint8_t link_id = prev_links[i];
|
|
int64_t miss_id = pkt_id - (i+1);
|
|
struct timing_entry *te = &lightc->historic[miss_id % HISTORIC_SIZE];
|
|
if (te->pkt_id == miss_id) continue; // Entry already exists
|
|
te->state = OOO_ONGOING;
|
|
te->detected_at = now;
|
|
te->link_id = link_id;
|
|
te->pkt_id = miss_id;
|
|
}
|
|
|
|
// Update current packet status
|
|
int link_id = url_get_port_int(fdinfo->url) - 7500;
|
|
struct timing_entry *te2 = &lightc->historic[pkt_id % HISTORIC_SIZE];
|
|
te2->state = te2->pkt_id == pkt_id ? OOO_DONE : IN_ORDER;
|
|
te2->pkt_id = pkt_id;
|
|
te2->link_id = link_id;
|
|
te2->finished_at = now;
|
|
}
|
|
|
|
int deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
|
|
char url[256];
|
|
struct evt_core_fdinfo *to_fdinfo = NULL;
|
|
struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
|
|
union abstract_packet* ap = (union abstract_packet*) &bp->ip;
|
|
|
|
if (ctx->verbose > 1) fprintf(stderr, " [algo_lightning] 1/2 Find destination\n");
|
|
sprintf(url, "udp:write:127.0.0.1:%d", ap->fmt.content.udp_encapsulated.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_wtof (&app_ctx->br, fdinfo);
|
|
return 1;
|
|
}
|
|
|
|
if (ctx->verbose > 1) fprintf(stderr, " [algo_lightning] 2/2 Move buffer\n");
|
|
mv_buffer_rtow (&app_ctx->br, fdinfo, to_fdinfo);
|
|
main_on_udp_write(ctx, to_fdinfo);
|
|
|
|
return 0;
|
|
}
|
|
|
|
int algo_lightning_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
|
|
monitoring(ctx, fdinfo, bp);
|
|
return deliver(ctx, fdinfo, bp);
|
|
}
|
|
|
|
void algo_lightning_update_stats (struct light_ctx *lightc, int64_t *stats) {
|
|
struct timespec now, not_before = {0}, temp_time;
|
|
set_now(&now);
|
|
timespec_diff (&now, &lightc->window, ¬_before);
|
|
|
|
for (int i = 0; i < lightc->total_links; i++) stats[i] = -1;
|
|
|
|
for (int i = 0; i < HISTORIC_SIZE; i++) {
|
|
if (timespec_lt(&lightc->historic[i].finished_at, ¬_before)) continue;
|
|
int64_t delta = 0;
|
|
switch (lightc->historic[i].state) {
|
|
case IN_ORDER:
|
|
break;
|
|
case OOO_ONGOING:
|
|
timespec_diff(&now, &lightc->historic[i].detected_at, &temp_time);
|
|
delta = timespec_get_unit (&temp_time, MILISEC);
|
|
break;
|
|
case OOO_DONE:
|
|
timespec_diff(&lightc->historic[i].finished_at, &lightc->historic[i].detected_at, &temp_time);
|
|
delta = timespec_get_unit (&temp_time, MILISEC);
|
|
break;
|
|
}
|
|
uint8_t l = lightc->historic[i].link_id;
|
|
stats[l] = delta > stats[l] ? delta : stats[l];
|
|
}
|
|
}
|
|
|
|
int send_message(struct evt_core_ctx* ctx, struct buffer_packet* bp) {
|
|
char url[256];
|
|
struct evt_core_cat *cat = evt_core_get_from_cat (ctx, "tcp-write");
|
|
if (cat == NULL) {
|
|
fprintf(stderr, "[algo_lightning] cat tcp-write not found\n");
|
|
exit(EXIT_FAILURE);
|
|
}
|
|
|
|
struct algo_ctx* app_ctx = cat->app_ctx;
|
|
struct light_ctx* lightc = app_ctx->misc;
|
|
set_now(&lightc->last[lightc->selected_link]);
|
|
|
|
sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + lightc->selected_link);
|
|
struct evt_core_fdinfo *to_fdinfo = evt_core_get_from_url (ctx, url);
|
|
if (to_fdinfo == NULL) return 0;
|
|
|
|
struct buffer_packet* bp_dup = dup_buffer_tow (&app_ctx->br, bp, to_fdinfo);
|
|
|
|
union abstract_packet monit = {
|
|
.fmt.headers.cmd = CMD_LINK_MONITORING_LIGHTNING,
|
|
.fmt.headers.size = lightc->monit_pkt_size,
|
|
.fmt.headers.flags = 0,
|
|
.fmt.content.link_monitoring_lightning.id = lightc->pkt_sent_id
|
|
};
|
|
union abstract_packet *ap_buf = buffer_append_ap (bp_dup, &monit);
|
|
uint8_t *links = &ap_buf->fmt.content.link_monitoring_lightning.prev_links;
|
|
for (int i = 0; i < lightc->sent_past_links; i++) {
|
|
links[i] = lightc->prev_links[(lightc->pkt_sent_id - (i + 1)) % MAX_LINKS];
|
|
}
|
|
|
|
lightc->prev_links[lightc->pkt_sent_id % MAX_LINKS] = lightc->selected_link;
|
|
lightc->pkt_sent_id++;
|
|
|
|
if (ctx->verbose > 1) {
|
|
dump_buffer_packet(bp_dup);
|
|
fprintf(stderr, " [algo_lightning] Will send this info\n");
|
|
}
|
|
main_on_tcp_write(ctx, to_fdinfo);
|
|
|
|
return 1;
|
|
}
|
|
|
|
int algo_lightning_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 light_ctx* lightc = app_ctx->misc;
|
|
union abstract_packet* ap = (union abstract_packet*) &bp->ip;
|
|
|
|
// Last step, send packet
|
|
int64_t stats[MAX_LINKS];
|
|
algo_lightning_update_stats(lightc, stats);
|
|
|
|
printf("retained stat: ");
|
|
for (int i = 0; i < lightc->total_links; i++) {
|
|
printf("%ld, ", stats[i]);
|
|
}
|
|
printf("\n");
|
|
|
|
struct timespec now, temp_time;
|
|
set_now(&now);
|
|
|
|
int protection = lightc->total_links;
|
|
while (protection-- > 0) {
|
|
lightc->selected_link = (lightc->selected_link + 1) % lightc->total_links;
|
|
int64_t cdelta = stats[lightc->selected_link];
|
|
|
|
// OK link
|
|
printf("cdelta=%ld\n", cdelta);
|
|
if (cdelta < lightc->max_ooo) {
|
|
send_message (ctx, bp);
|
|
break;
|
|
}
|
|
|
|
// Broken link that must be probed
|
|
timespec_diff (&now, &lightc->last[lightc->selected_link], &temp_time);
|
|
uint64_t elapsed = timespec_get_unit(&temp_time, MILISEC);
|
|
if (elapsed >= lightc->sleep_duration) {
|
|
send_message (ctx, bp);
|
|
continue;
|
|
}
|
|
|
|
// Broken link probed recently
|
|
}
|
|
if (protection <= 0) {
|
|
fprintf(stderr, "Schedule error\n");
|
|
exit(EXIT_FAILURE);
|
|
}
|
|
|
|
mv_buffer_rtof (&app_ctx->br, fdinfo);
|
|
return 0;
|
|
}
|
|
|
|
int algo_lightning_on_err(struct evt_core_ctx *ctx, struct evt_core_fdinfo *fdinfo) {
|
|
return 0;
|
|
}
|