diff --git a/src/algo_thunder.c b/src/algo_thunder.c index 0aa3e03..faaedb0 100644 --- a/src/algo_thunder.c +++ b/src/algo_thunder.c @@ -8,11 +8,11 @@ struct thunder_ctx { uint16_t recv_id; uint16_t emit_id; - uint64_t lid_sent; uint8_t selected_link; uint8_t total_links; - uint8_t lid_sent_per_link[64]; uint8_t delta_t_per_link[64]; + uint8_t blacklisted[64]; + struct timespec prev_link_time; }; @@ -26,7 +26,7 @@ void prepare(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu .fmt.headers.size = sizeof(metadata.fmt.headers) + sizeof(metadata.fmt.content.udp_metadata_thunder), .fmt.headers.flags = 0, .fmt.content.udp_metadata_thunder.id = thunderc->emit_id, - .fmt.content.udp_metadata_thunder.deltat = 0 //@FIXME + .fmt.content.udp_metadata_thunder.deltat = 0 //@FIXME delta t must be set }; buffer_append_ap (bp, &metadata); @@ -42,32 +42,50 @@ int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu struct evt_core_fdinfo *to_fdinfo = NULL; struct evt_core_cat* cat = evt_core_get_from_cat (ctx, "tcp-write"); + struct timespec curr; + int secs, nsecs; + uint64_t mili_sec; + + do { - thunderc->selected_link = thunderc->selected_link + 1 % cat->socklist->len; - - if (thunderc->selected_link == 0) { - thunderc->lid_sent++; - for (int i = 0; i < thunderc->total_links; i++) { - thunderc->lid_sent_per_link[i]++; - } - } + // 1. We choose the link + thunderc->selected_link = (thunderc->selected_link + 1) % cat->socklist->len; + // 2. We create the packet template union abstract_packet links = { .fmt.headers.cmd = CMD_LINK_MONITORING_THUNDER, - .fmt.headers.size = sizeof(links.fmt.headers) + sizeof(links.fmt.content.link_monitoring_thunder), + .fmt.headers.size = sizeof(links.fmt.headers) + sizeof(links.fmt.content.link_monitoring_thunder) + sizeof(struct link_info) * (thunderc->total_links - 1), .fmt.headers.flags = 0, - .fmt.content.link_monitoring_thunder.lid = thunderc->lid_sent, - .fmt.content.link_monitoring_thunder.links_status = NULL + .fmt.content.link_monitoring_thunder.links_status = {} }; to_fdinfo = g_array_index(cat->socklist, struct evt_core_fdinfo*, thunderc->selected_link); - // We move the buffer and notify the target + // 3. We append the template to the buffer struct buffer_packet* bp_dup = dup_buffer_tow (&app_ctx->br, bp, to_fdinfo); - buffer_append_ap (bp_dup, &links); + union abstract_packet *new_ap = buffer_append_ap (bp_dup, &links); + + // 4. We compute the time difference + if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){ + perror("clock_gettime error"); + exit(EXIT_FAILURE); + } + secs = curr.tv_sec - thunderc->prev_link_time.tv_sec; + nsecs = curr.tv_nsec - thunderc->prev_link_time.tv_nsec; + thunderc->prev_link_time = curr; + mili_sec = secs * 1000 + nsecs / 1000000; + if (mili_sec > 200) mili_sec = 200; + + // 5. We create the array + struct link_info *li = &new_ap->fmt.content.link_monitoring_thunder.links_status; + for (int i = 0; i < thunderc->total_links; i++) { + thunderc->delta_t_per_link[i] += mili_sec; + li[i].delta_t = thunderc->delta_t_per_link[i]; + } + li[thunderc->selected_link].delta_t = 0; main_on_tcp_write(ctx, to_fdinfo); - } while (1); + } while (thunderc->blacklisted[thunderc->selected_link]); if (ctx->verbose > 1) fprintf(stderr, " [algo_thunder] Packets sent\n"); @@ -85,7 +103,7 @@ void algo_thunder_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, struc } memset(app_ctx->misc, 0, sizeof(struct thunder_ctx)); struct thunder_ctx* thunderc = app_ctx->misc; - + thunderc->selected_link = UINT8_MAX - 1; } int algo_thunder_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { diff --git a/src/packet.c b/src/packet.c index c1e845b..85989a7 100644 --- a/src/packet.c +++ b/src/packet.c @@ -32,10 +32,12 @@ size_t get_full_size(struct buffer_packet* bp) { return &(buffer_free_ap (bp))->raw - &bp->ip[0]; } -void buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap) { +union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap) { buffer_last_ap(bp)->fmt.headers.flags |= FLAG_READ_NEXT; - memcpy(buffer_last_ap(bp), ap, ap->fmt.headers.size); + union abstract_packet *new_ap = buffer_last_ap(bp); + memcpy(new_ap, ap, ap->fmt.headers.size); bp->ap_count++; + return new_ap; } enum FD_STATE read_packet_from_tcp(struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { diff --git a/src/packet.h b/src/packet.h index 1c1ffdb..6b1530f 100644 --- a/src/packet.h +++ b/src/packet.h @@ -43,7 +43,6 @@ enum PKT_FLAGS { struct link_info { uint8_t delta_t; - uint8_t delta_lid; }; union abstract_packet { @@ -57,8 +56,7 @@ union abstract_packet { union { struct { - uint16_t lid; - struct link_info *links_status; + struct link_info links_status; } link_monitoring_thunder; struct { uint16_t id; @@ -90,7 +88,7 @@ struct udp_target { size_t get_full_size(struct buffer_packet* bp); -void buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap); +union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap); union abstract_packet* buffer_free_ptr(struct buffer_packet* bp); union abstract_packet* buffer_last_ptr(struct buffer_packet* bp); union abstract_packet* ap_next(union abstract_packet* ap);