From 67f8182ef9d64f0c2cb94ce8229a78fd29a85281 Mon Sep 17 00:00:00 2001 From: Christian Giese Date: Mon, 29 Mar 2021 22:46:48 +0200 Subject: [PATCH] Traffic Streams (WIP) --- src/bbl.c | 9 +- src/bbl.h | 20 +- src/bbl_config.c | 120 +++++++- src/bbl_ctrl.c | 67 ++++ src/bbl_ctx.c | 43 +-- src/bbl_io_netmap.c | 26 +- src/bbl_io_netmap.h | 3 + src/bbl_io_packet_mmap.c | 69 ++++- src/bbl_io_packet_mmap.h | 3 + src/bbl_io_raw.c | 37 ++- src/bbl_io_raw.h | 3 + src/bbl_protocols.c | 14 +- src/bbl_protocols.h | 6 +- src/bbl_rx.c | 305 +++++++++++-------- src/bbl_session.c | 7 + src/bbl_stats.h | 1 + src/bbl_stream.c | 638 +++++++++++++++++++++++++++++++++++++++ src/bbl_stream.h | 92 ++++++ src/bbl_timer.c | 14 +- src/bbl_utils.h | 1 - 20 files changed, 1278 insertions(+), 200 deletions(-) create mode 100644 src/bbl_stream.c create mode 100644 src/bbl_stream.h diff --git a/src/bbl.c b/src/bbl.c index e6aa554e..73240907 100644 --- a/src/bbl.c +++ b/src/bbl.c @@ -37,7 +37,10 @@ volatile uint8_t g_teardown_request_count = 0; /* This global variable is used to switch between access * interfaces in interactive mode (ncurses). */ -uint8_t g_access_if_selected = 0; +uint8_t g_access_if_selected = 0; + +uint8_t g_display_streams = 0; +uint8_t g_stream_start_index = 0; const char banner[] = "\n" " ____ __ ____ _ __ ,/\n" @@ -358,7 +361,7 @@ bbl_add_access_interfaces (bbl_ctx_s *ctx) { } } } - access_if = bbl_add_interface(ctx, access_config->interface, 1024); + access_if = bbl_add_interface(ctx, access_config->interface, ctx->config.io_slots); if (!access_if) { LOG(ERROR, "Failed to add access interface %s\n", access_config->interface); return false; @@ -738,7 +741,7 @@ main (int argc, char *argv[]) * Add network interface. */ if (strlen(ctx->config.network_if)) { - ctx->op.network_if = bbl_add_interface(ctx, ctx->config.network_if, 1024); + ctx->op.network_if = bbl_add_interface(ctx, ctx->config.network_if, ctx->config.io_slots); if (!ctx->op.network_if) { if (interactive) endwin(); fprintf(stderr, "Error: Failed to add network interface\n"); diff --git a/src/bbl.h b/src/bbl.h index 28e8cd58..38bb178f 100644 --- a/src/bbl.h +++ b/src/bbl.h @@ -86,7 +86,7 @@ #define BBL_MAX_ACCESS_INTERFACES 64 #define BBL_AVG_SAMPLES 5 -#define DATA_TRAFFIC_MAX_LEN 1500 +#define DATA_TRAFFIC_MAX_LEN 2000 #define UNUSED(x) (void)x @@ -184,6 +184,8 @@ typedef struct bbl_interface_ struct { uint64_t packets_tx; uint64_t packets_rx; + uint64_t bytes_tx; + uint64_t bytes_rx; bbl_rate_s rate_packets_tx; bbl_rate_s rate_packets_rx; uint64_t packets_rx_drop_unknown; @@ -282,6 +284,7 @@ typedef struct bbl_interface_ struct timespec tx_timestamp; /* user space timestamps */ struct timespec rx_timestamp; /* user space timestamps */ + CIRCLEQ_HEAD(bbl_interface__, bbl_session_ ) session_tx_qhead; /* list of sessions that want to transmit */ CIRCLEQ_HEAD(bbl_interface___, bbl_l2tp_queue_ ) l2tp_tx_qhead; /* list of messages that want to transmit */ } bbl_interface_s; @@ -297,6 +300,8 @@ typedef struct bbl_access_config_ bbl_access_type_t access_type; /* pppoe or ipoe */ bbl_vlan_mode_t vlan_mode; /* 1:1 (default) or N:1 */ + uint16_t stream_group_id; + uint16_t access_outer_vlan; uint16_t access_outer_vlan_min; uint16_t access_outer_vlan_max; @@ -381,6 +386,7 @@ typedef struct bbl_ctx_ dict *vlan_session_dict; /* hashtable for 1:1 vlan sessions */ dict *l2tp_session_dict; /* hashtable for L2TP sessions */ dict *li_flow_dict; /* hashtable for LI flows */ + dict *stream_flow_dict; /* hashtable for traffic stream flows */ uint16_t next_tunnel_id; @@ -432,9 +438,11 @@ typedef struct bbl_ctx_ struct { bool interface_lock_force; - uint16_t tx_interval; - uint16_t rx_interval; + uint64_t tx_interval; /* TX interval in nsec */ + uint64_t rx_interval; /* RX interval in nsec */ + uint16_t io_slots; + bool qdisc_bypass; bbl_io_mode_t io_mode; @@ -453,6 +461,9 @@ typedef struct bbl_ctx_ /* Access Interfaces */ bbl_access_config_s *access_config; + /* Traffic Streams */ + void *stream_config; + /* Global Session Settings */ uint32_t sessions; uint32_t sessions_max_outstanding; @@ -598,6 +609,7 @@ typedef struct vlan_session_key_ { #define BBL_SESSION_HASHTABLE_SIZE 128993 /* is a prime number */ #define BBL_LI_HASHTABLE_SIZE 32771 /* is a prime number */ +#define BBL_STREAM_FLOW_HASHTABLE_SIZE 32771 /* is a prime number */ /* * Client Session to a BNG device. @@ -641,6 +653,8 @@ typedef struct bbl_session_ bbl_access_type_t access_type; + uint16_t stream_group_id; + void *stream; struct { uint32_t ifindex; uint16_t outer_vlan_id; diff --git a/src/bbl_config.c b/src/bbl_config.c index 2c9dfa41..d5fa7ce3 100644 --- a/src/bbl_config.c +++ b/src/bbl_config.c @@ -9,6 +9,7 @@ #include "bbl.h" #include "bbl_config.h" +#include "bbl_stream.h" #include #include @@ -230,6 +231,88 @@ json_parse_access_interface (bbl_ctx_s *ctx, json_t *access_interface, bbl_acces } else { access_config->session_traffic_autostart = ctx->config.session_traffic_autostart; } + + value = json_object_get(access_interface, "stream-group-id"); + if (value) { + access_config->stream_group_id = json_number_value(value); + } + return true; +} + +static bool +json_parse_stream (json_t *stream, bbl_stream_config *stream_config) { + json_t *value = NULL; + const char *s = NULL; + + if (json_unpack(stream, "{s:s}", "type", &s) == 0) { + if (strcmp(s, "ipv4") == 0) { + stream_config->type = STREAM_IPV4; + } else if (strcmp(s, "ipv6") == 0) { + stream_config->type = STREAM_IPV6; + } else if (strcmp(s, "ipv6pd") == 0) { + stream_config->type = STREAM_IPV6PD; + } else if (strcmp(s, "l2tp") == 0) { + stream_config->type = STREAM_L2TP; + } else { + fprintf(stderr, "JSON config error: Invalid value for stream->type\n"); + return false; + } + } else { + fprintf(stderr, "JSON config error: Missing value for stream->type\n"); + return false; + } + + if (json_unpack(stream, "{s:s}", "direction", &s) == 0) { + if (strcmp(s, "upstream") == 0) { + stream_config->direction = STREAM_DIRECTION_UP; + } else if (strcmp(s, "downstream") == 0) { + stream_config->direction = STREAM_DIRECTION_DOWN; + } else if (strcmp(s, "both") == 0) { + stream_config->direction = STREAM_DIRECTION_BOTH; + } else { + fprintf(stderr, "JSON config error: Invalid value for stream->direction\n"); + return false; + } + } else { + stream_config->direction = STREAM_DIRECTION_BOTH; + return false; + } + + if (json_unpack(stream, "{s:s}", "name", &s) == 0) { + stream_config->name = strdup(s); + } else { + fprintf(stderr, "JSON config error: Missing value for stream->name\n"); + return false; + } + + value = json_object_get(stream, "stream-group-id"); + if (value) { + stream_config->stream_group_id = json_number_value(value); + } + + value = json_object_get(stream, "length"); + if (value) { + stream_config->length = json_number_value(value); + } else { + stream_config->length = 128; + } + + value = json_object_get(stream, "priority"); + if (value) { + stream_config->priority = json_number_value(value); + } + + value = json_object_get(stream, "vlan-priority"); + if (value) { + stream_config->vlan_priority = json_number_value(value); + } + + value = json_object_get(stream, "pps"); + if (value) { + stream_config->pps = json_number_value(value); + } else { + stream_config->pps = 1; + } return true; } @@ -240,9 +323,10 @@ json_parse_config (json_t *root, bbl_ctx_s *ctx) { const char *s; uint32_t ipv4; int i, size; - bbl_access_config_s *access_config = NULL; - bbl_l2tp_server_t *l2tp_server = NULL; - bbl_secondary_ip_s *secondary_ip; + bbl_access_config_s *access_config = NULL; + bbl_stream_config *stream_config = NULL; + bbl_l2tp_server_t *l2tp_server = NULL; + bbl_secondary_ip_s *secondary_ip; if(json_typeof(root) != JSON_OBJECT) { fprintf(stderr, "JSON config error: Configuration root element must object\n"); @@ -593,17 +677,16 @@ json_parse_config (json_t *root, bbl_ctx_s *ctx) { } } - /* Interface Configuration */ section = json_object_get(root, "interfaces"); if (json_is_object(section)) { value = json_object_get(section, "tx-interval"); if (json_is_number(value)) { - ctx->config.tx_interval = json_number_value(value); + ctx->config.tx_interval = json_number_value(value) * MSEC; } value = json_object_get(section, "rx-interval"); if (json_is_number(value)) { - ctx->config.rx_interval = json_number_value(value); + ctx->config.rx_interval = json_number_value(value) * MSEC; } value = json_object_get(section, "qdisc-bypass"); if (json_is_boolean(value)) { @@ -814,6 +897,26 @@ json_parse_config (json_t *root, bbl_ctx_s *ctx) { fprintf(stderr, "JSON config error: List expected in L2TP server configuration but dictionary found\n"); } + /* Traffic Streams Configuration */ + section = json_object_get(root, "streams"); + if (json_is_array(section)) { + /* Config is provided as array (multiple streams) */ + size = json_array_size(section); + for (i = 0; i < size; i++) { + if(!stream_config) { + ctx->config.stream_config = malloc(sizeof(bbl_stream_config)); + stream_config = ctx->config.stream_config; + } else { + stream_config->next = malloc(sizeof(bbl_stream_config)); + stream_config = stream_config->next; + } + memset(stream_config, 0x0, sizeof(bbl_stream_config)); + if(!json_parse_stream(json_array_get(section, i), stream_config)) { + return false; + } + } + } + return true; } @@ -851,8 +954,9 @@ bbl_config_init_defaults (bbl_ctx_s *ctx) { ctx->config.password = (char *)g_default_pass; ctx->config.agent_remote_id = (char *)g_default_ari; ctx->config.agent_circuit_id = (char *)g_default_aci; - ctx->config.tx_interval = 5; - ctx->config.rx_interval = 5; + ctx->config.tx_interval = 5 * MSEC; + ctx->config.rx_interval = 5 * MSEC; + ctx->config.io_slots = 1024; ctx->config.qdisc_bypass = true; ctx->config.sessions = 1; ctx->config.sessions_max_outstanding = 800; diff --git a/src/bbl_ctrl.c b/src/bbl_ctrl.c index 8c532d9d..3879d85a 100644 --- a/src/bbl_ctrl.c +++ b/src/bbl_ctrl.c @@ -24,6 +24,7 @@ #include "bbl_ctrl.h" #include "bbl_logging.h" #include "bbl_session.h" +#include "bbl_stream.h" #define BACKLOG 4 #define INPUT_BUFFER 1024 @@ -972,6 +973,71 @@ bbl_ctrl_l2tp_csurq(int fd, bbl_ctx_s *ctx, uint32_t session_id __attribute__((u } } +ssize_t +bbl_ctrl_session_streams(int fd, bbl_ctx_s *ctx, uint32_t session_id, json_t* arguments __attribute__((unused))) { + ssize_t result = 0; + json_t *root; + json_t *json_streams = NULL; + json_t *json_stream = NULL; + + bbl_session_s *session; + bbl_stream *stream; + + if(session_id == 0) { + /* session-id is mandatory */ + return bbl_ctrl_status(fd, "error", 400, "missing session-id"); + } + + session = bbl_session_get(ctx, session_id); + if(session) { + stream = session->stream; + + json_streams = json_array(); + while(stream) { + json_stream = json_pack("{ss ss si si si si si si si si si si si si si si si si si si}", + "name", stream->config->name, + "direction", stream->direction == STREAM_DIRECTION_UP ? "upstream" : "dowstream", + "flow-id", stream->flow_id, + "rx-first-seq", stream->rx_first_seq, + "rx-last-seq", stream->rx_last_seq, + "rx-tos-tc", stream->rx_priority, + "rx-outer-vlan-pbit", stream->rx_outer_vlan_pbit, + "rx-inner-vlan-pbit", stream->rx_inner_vlan_pbit, + "rx-len", stream->rx_len, + "tx-len", stream->tx_len, + "rx-packets", stream->packets_rx, + "tx-packets", stream->packets_tx, + "rx-loss", stream->loss, + "rx-delay-nsec-min", stream->min_delay_ns, + "rx-delay-nsec-max", stream->max_delay_ns, + "rx-pps", stream->rate_packets_rx.avg, + "tx-pps", stream->rate_packets_tx.avg, + "tx-bps-l2", stream->rate_packets_tx.avg * stream->tx_len * 8, + "rx-bps-l2", stream->rate_packets_rx.avg * stream->rx_len * 8, + "rx-bps-l3", stream->rate_packets_rx.avg * stream->config->length * 8); + + json_array_append(json_streams, json_stream); + stream = stream->next; + } + root = json_pack("{ss si s{si so*}}", + "status", "ok", + "code", 200, + "session-streams", + "session-id", session->session_id, + "streams", json_streams); + if(root) { + result = json_dumpfd(root, fd, 0); + json_decref(root); + } else { + result = bbl_ctrl_status(fd, "error", 500, "internal error"); + json_decref(json_streams); + } + return result; + } else { + return bbl_ctrl_status(fd, "warning", 404, "session not found"); + } +} + struct action { char *name; callback_function *fn; @@ -1000,6 +1066,7 @@ struct action actions[] = { {"l2tp-tunnels", bbl_ctrl_l2tp_tunnels}, {"l2tp-sessions", bbl_ctrl_l2tp_sessions}, {"l2tp-csurq", bbl_ctrl_l2tp_csurq}, + {"session-streams", bbl_ctrl_session_streams}, {NULL, NULL}, }; diff --git a/src/bbl_ctx.c b/src/bbl_ctx.c index 94493beb..a57496af 100644 --- a/src/bbl_ctx.c +++ b/src/bbl_ctx.c @@ -11,26 +11,6 @@ extern volatile bool g_teardown; -int -bbl_compare_session (void *key1, void *key2) -{ - const uint64_t a = *(const uint64_t*)key1; - const uint64_t b = *(const uint64_t*)key2; - return (a > b) - (a < b); -} - -uint -bbl_session_hash (const void* k) -{ - uint hash = 2166136261U; - - hash ^= *(uint32_t *)k; - hash ^= *(uint16_t *)(k+4) << 12; - hash ^= *(uint16_t *)(k+6); - - return hash; -} - int bbl_compare_key32 (void *key1, void *key2) { @@ -47,6 +27,26 @@ bbl_key32_hash (const void* k) return hash; } +int +bbl_compare_key64 (void *key1, void *key2) +{ + const uint64_t a = *(const uint64_t*)key1; + const uint64_t b = *(const uint64_t*)key2; + return (a > b) - (a < b); +} + +uint +bbl_key64_hash (const void* k) +{ + uint hash = 2166136261U; + + hash ^= *(uint32_t *)k; + hash ^= *(uint16_t *)(k+4) << 12; + hash ^= *(uint16_t *)(k+6); + + return hash; +} + /** * bbl_ctx_add * @@ -78,9 +78,10 @@ bbl_ctx_add (void) ctx->flow_id = 1; /* Initialize hash table dictionaries. */ - ctx->vlan_session_dict = hashtable2_dict_new((dict_compare_func)bbl_compare_session, bbl_session_hash, BBL_SESSION_HASHTABLE_SIZE); + ctx->vlan_session_dict = hashtable2_dict_new((dict_compare_func)bbl_compare_key64, bbl_key64_hash, BBL_SESSION_HASHTABLE_SIZE); ctx->l2tp_session_dict = hashtable2_dict_new((dict_compare_func)bbl_compare_key32, bbl_key32_hash, BBL_SESSION_HASHTABLE_SIZE); ctx->li_flow_dict = hashtable2_dict_new((dict_compare_func)bbl_compare_key32, bbl_key32_hash, BBL_LI_HASHTABLE_SIZE); + ctx->stream_flow_dict = hashtable2_dict_new((dict_compare_func)bbl_compare_key64, bbl_key64_hash, BBL_STREAM_FLOW_HASHTABLE_SIZE); return ctx; } diff --git a/src/bbl_io_netmap.c b/src/bbl_io_netmap.c index c765769c..1671b277 100644 --- a/src/bbl_io_netmap.c +++ b/src/bbl_io_netmap.c @@ -47,6 +47,7 @@ bbl_io_netmap_rx_job (timer_s *timer) eth_start = (uint8_t*)NETMAP_BUF(ring, ring->slot[i].buf_idx); eth_len = ring->slot[i].len; interface->stats.packets_rx++; + interface->stats.bytes_rx += eth_len; /* * Dump the packet into pcap file. @@ -59,8 +60,8 @@ bbl_io_netmap_rx_job (timer_s *timer) decode_result = decode_ethernet(eth_start, eth_len, interface->ctx->sp_rx, SCRATCHPAD_LEN, ð); if(decode_result == PROTOCOL_SUCCESS) { /* Copy RX timestamp */ - eth->rx_sec = ring->ts.tv_sec; - eth->rx_nsec = ring->ts.tv_usec * 1000; + eth->timestamp.tv_sec = ring->ts.tv_sec; + eth->timestamp.tv_nsec = ring->ts.tv_usec * 1000; if(interface->access) { bbl_rx_handler_access(eth, interface); } else { @@ -118,6 +119,7 @@ bbl_io_netmap_tx_job (timer_s *timer) if (tx_result == PROTOCOL_SUCCESS) { send = true; interface->stats.packets_tx++; + interface->stats.bytes_tx += len; ring->slot[i].len = len; ring->head = ring->cur = nm_ring_next(ring, i); /* Dump the packet into pcap file. */ @@ -134,6 +136,22 @@ bbl_io_netmap_tx_job (timer_s *timer) } } +/** + * bbl_io_netmap_send + * + * @param interface interface. + * @param packet packet to be send + * @param packet_len packet length + */ +bool +bbl_io_netmap_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len) { + /* NOT IMPLEMENTED */ + UNUSED(interface); + UNUSED(packet); + UNUSED(packet_len); + return false; +} + /** * bbl_io_netmap_add_interface * @@ -172,9 +190,9 @@ bbl_io_netmap_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int slot * Add an periodic timer for polling I/O. */ snprintf(timer_name, sizeof(timer_name), "%s TX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval * MSEC, interface, bbl_io_netmap_tx_job); + timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval, interface, bbl_io_netmap_tx_job); snprintf(timer_name, sizeof(timer_name), "%s RX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval * MSEC, interface, bbl_io_netmap_rx_job); + timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval, interface, bbl_io_netmap_rx_job); return true; } diff --git a/src/bbl_io_netmap.h b/src/bbl_io_netmap.h index 51282a3d..3d7c50e9 100644 --- a/src/bbl_io_netmap.h +++ b/src/bbl_io_netmap.h @@ -22,6 +22,9 @@ typedef struct bbl_io_netmap_ctx_ struct nm_desc *port; } bbl_io_netmap_ctx; +bool +bbl_io_netmap_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len); + bool bbl_io_netmap_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int slots); diff --git a/src/bbl_io_packet_mmap.c b/src/bbl_io_packet_mmap.c index e2aee422..66e5ff8b 100644 --- a/src/bbl_io_packet_mmap.c +++ b/src/bbl_io_packet_mmap.c @@ -13,8 +13,7 @@ #include "bbl_tx.h" void -bbl_io_packet_mmap_rx_job (timer_s *timer) -{ +bbl_io_packet_mmap_rx_job (timer_s *timer) { bbl_interface_s *interface; bbl_ctx_s *ctx; bbl_io_packet_mmap_ctx *io_ctx; @@ -58,6 +57,7 @@ bbl_io_packet_mmap_rx_job (timer_s *timer) eth_start = (uint8_t*)tphdr + tphdr->tp_mac; eth_len = tphdr->tp_len; interface->stats.packets_rx++; + interface->stats.bytes_rx += eth_len; /* * Dump the packet into pcap file. @@ -75,8 +75,8 @@ bbl_io_packet_mmap_rx_job (timer_s *timer) eth->vlan_outer = tphdr->tp_vlan_tci & ETH_VLAN_ID_MAX; #endif /* Copy RX timestamp */ - eth->rx_sec = tphdr->tp_sec; /* ktime/hw timestamp */ - eth->rx_nsec = tphdr->tp_nsec; /* ktime/hw timestamp */ + eth->timestamp.tv_sec = tphdr->tp_sec; /* ktime/hw timestamp */ + eth->timestamp.tv_nsec = tphdr->tp_nsec; /* ktime/hw timestamp */ if(interface->access) { bbl_rx_handler_access(eth, interface); } else { @@ -98,8 +98,7 @@ bbl_io_packet_mmap_rx_job (timer_s *timer) } void -bbl_io_packet_mmap_tx_job (timer_s *timer) -{ +bbl_io_packet_mmap_tx_job (timer_s *timer) { bbl_interface_s *interface; bbl_ctx_s *ctx; bbl_io_packet_mmap_ctx *io_ctx; @@ -150,6 +149,7 @@ bbl_io_packet_mmap_tx_job (timer_s *timer) tx_result = bbl_tx(ctx, interface, buf, &len); if (tx_result == PROTOCOL_SUCCESS) { interface->stats.packets_tx++; + interface->stats.bytes_tx += len; tphdr->tp_len = len; tphdr->tp_status = TP_STATUS_SEND_REQUEST; io_ctx->cursor_tx = (io_ctx->cursor_tx + 1) % io_ctx->req_tx.tp_frame_nr; @@ -174,6 +174,59 @@ bbl_io_packet_mmap_tx_job (timer_s *timer) } } +/** + * bbl_io_packet_mmap_send + * + * @param interface interface. + * @param packet packet to be send + * @param packet_len packet length + */ +bool +bbl_io_packet_mmap_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len) { + bbl_ctx_s *ctx; + bbl_io_packet_mmap_ctx *io_ctx; + + struct tpacket2_hdr* tphdr; + + uint8_t *frame_ptr; + + ctx = interface->ctx; + io_ctx = interface->io_ctx; + + frame_ptr = io_ctx->ring_tx + (io_ctx->cursor_tx * io_ctx->req_tx.tp_frame_size); + tphdr = (struct tpacket2_hdr *)frame_ptr; + + if (tphdr->tp_status != TP_STATUS_AVAILABLE) { + interface->stats.no_tx_buffer++; + return false; + } + + memcpy(frame_ptr + TPACKET2_HDRLEN - sizeof(struct sockaddr_ll), packet, packet_len); + interface->stats.packets_tx++; + interface->stats.bytes_tx += packet_len; + tphdr->tp_len = packet_len; + tphdr->tp_status = TP_STATUS_SEND_REQUEST; + io_ctx->cursor_tx = (io_ctx->cursor_tx + 1) % io_ctx->req_tx.tp_frame_nr; + /* Dump the packet into pcap file. */ + if (ctx->pcap.write_buf) { + pcapng_push_packet_header(ctx, &interface->tx_timestamp, + packet, packet_len, interface->pcap_index, + PCAPNG_EPB_FLAGS_OUTBOUND); + pcapng_fflush(ctx); + } + +#if 0 + /* Notify kernel. */ + if (sendto(io_ctx->fd_tx, NULL, 0 , 0, NULL, 0) == -1) { + LOG(IO, "Sendto failed with errno: %i\n", errno); + interface->stats.sendto_failed++; + return false; + } +#endif + + return true; +} + /** * bbl_io_packet_mmap_add_interface * @@ -326,9 +379,9 @@ bbl_io_packet_mmap_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int * Add an periodic timer for polling I/O. */ snprintf(timer_name, sizeof(timer_name), "%s TX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval * MSEC, interface, bbl_io_packet_mmap_tx_job); + timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval, interface, bbl_io_packet_mmap_tx_job); snprintf(timer_name, sizeof(timer_name), "%s RX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval * MSEC, interface, bbl_io_packet_mmap_rx_job); + timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval, interface, bbl_io_packet_mmap_rx_job); return true; } \ No newline at end of file diff --git a/src/bbl_io_packet_mmap.h b/src/bbl_io_packet_mmap.h index 034ba483..2f7da2ab 100644 --- a/src/bbl_io_packet_mmap.h +++ b/src/bbl_io_packet_mmap.h @@ -34,6 +34,9 @@ typedef struct bbl_io_packet_mmap_ctx_ } bbl_io_packet_mmap_ctx; +bool +bbl_io_packet_mmap_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len); + bool bbl_io_packet_mmap_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int slots); diff --git a/src/bbl_io_raw.c b/src/bbl_io_raw.c index 5d0cb4e7..93926e8e 100644 --- a/src/bbl_io_raw.c +++ b/src/bbl_io_raw.c @@ -13,8 +13,7 @@ #include "bbl_tx.h" void -bbl_io_raw_rx_job (timer_s *timer) -{ +bbl_io_raw_rx_job (timer_s *timer) { bbl_interface_s *interface; bbl_ctx_s *ctx; bbl_io_raw_ctx *io_ctx; @@ -44,10 +43,12 @@ bbl_io_raw_rx_job (timer_s *timer) if(rx_len < 14) { break; } - interface->stats.packets_rx++; eth_start = io_ctx->buf; eth_len = rx_len; - + + interface->stats.packets_rx++; + interface->stats.bytes_rx += eth_len; + /* * Dump the packet into pcap file. */ @@ -59,8 +60,8 @@ bbl_io_raw_rx_job (timer_s *timer) decode_result = decode_ethernet(eth_start, eth_len, interface->ctx->sp_rx, SCRATCHPAD_LEN, ð); if(decode_result == PROTOCOL_SUCCESS) { /* Copy RX timestamp */ - eth->rx_sec = interface->rx_timestamp.tv_sec; - eth->rx_nsec = interface->rx_timestamp.tv_nsec; + eth->timestamp.tv_sec = interface->rx_timestamp.tv_sec; + eth->timestamp.tv_nsec = interface->rx_timestamp.tv_nsec; if(interface->access) { bbl_rx_handler_access(eth, interface); } else { @@ -76,8 +77,7 @@ bbl_io_raw_rx_job (timer_s *timer) } void -bbl_io_raw_tx_job (timer_s *timer) -{ +bbl_io_raw_tx_job (timer_s *timer) { bbl_interface_s *interface; bbl_ctx_s *ctx; bbl_io_raw_ctx *io_ctx; @@ -106,6 +106,7 @@ bbl_io_raw_tx_job (timer_s *timer) return; } interface->stats.packets_tx++; + interface->stats.bytes_tx += len; /* Dump the packet into pcap file. */ if (ctx->pcap.write_buf) { pcapng_push_packet_header(ctx, &interface->tx_timestamp, @@ -118,6 +119,22 @@ bbl_io_raw_tx_job (timer_s *timer) pcapng_fflush(ctx); } +/** + * bbl_io_raw_send + * + * @param interface interface. + * @param packet packet to be send + * @param packet_len packet length + */ +bool +bbl_io_raw_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len) { + /* NOT IMPLEMENTED */ + UNUSED(interface); + UNUSED(packet); + UNUSED(packet_len); + return false; +} + /** * bbl_io_raw_add_interface * @@ -222,9 +239,9 @@ bbl_io_raw_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int slots) * Add an periodic timer for polling I/O. */ snprintf(timer_name, sizeof(timer_name), "%s TX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval * MSEC, interface, bbl_io_raw_tx_job); + timer_add_periodic(&ctx->timer_root, &interface->tx_job, timer_name, 0, ctx->config.tx_interval, interface, bbl_io_raw_tx_job); snprintf(timer_name, sizeof(timer_name), "%s RX", interface->name); - timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval * MSEC, interface, bbl_io_raw_rx_job); + timer_add_periodic(&ctx->timer_root, &interface->rx_job, timer_name, 0, ctx->config.rx_interval, interface, bbl_io_raw_rx_job); return true; } \ No newline at end of file diff --git a/src/bbl_io_raw.h b/src/bbl_io_raw.h index b00e7a0c..5d58d1c0 100644 --- a/src/bbl_io_raw.h +++ b/src/bbl_io_raw.h @@ -19,6 +19,9 @@ typedef struct bbl_io_raw_ctx_ uint8_t *buf; } bbl_io_raw_ctx; +bool +bbl_io_raw_send (bbl_interface_s *interface, uint8_t *packet, uint16_t packet_len); + bool bbl_io_raw_add_interface(bbl_ctx_s *ctx, bbl_interface_s *interface, int slots); diff --git a/src/bbl_protocols.c b/src/bbl_protocols.c index a58befef..d140de10 100644 --- a/src/bbl_protocols.c +++ b/src/bbl_protocols.c @@ -224,8 +224,10 @@ encode_bbl(uint8_t *buf, uint16_t *len, BUMP_WRITE_BUFFER(buf, len, sizeof(uint64_t)); *(uint64_t*)buf = bbl->flow_seq; BUMP_WRITE_BUFFER(buf, len, sizeof(uint64_t)); - *(uint64_t*)buf = bbl->timestamp; - BUMP_WRITE_BUFFER(buf, len, sizeof(uint64_t)); + *(uint32_t*)buf = bbl->timestamp.tv_sec; + BUMP_WRITE_BUFFER(buf, len, sizeof(uint32_t)); + *(uint32_t*)buf = bbl->timestamp.tv_nsec; + BUMP_WRITE_BUFFER(buf, len, sizeof(uint32_t)); return PROTOCOL_SUCCESS; } @@ -1536,8 +1538,10 @@ decode_bbl(uint8_t *buf, uint16_t len, BUMP_BUFFER(buf, len, sizeof(uint64_t)); bbl->flow_seq = *(uint64_t*)buf; BUMP_BUFFER(buf, len, sizeof(uint64_t)); - bbl->timestamp = *(uint64_t*)buf; - BUMP_BUFFER(buf, len, sizeof(uint64_t)); + bbl->timestamp.tv_sec = *(uint32_t*)buf; + BUMP_BUFFER(buf, len, sizeof(uint32_t)); + bbl->timestamp.tv_nsec = *(uint32_t*)buf; + BUMP_BUFFER(buf, len, sizeof(uint32_t)); *_bbl = bbl; return PROTOCOL_SUCCESS; @@ -2515,6 +2519,8 @@ decode_ethernet(uint8_t *buf, uint16_t len, memset(eth, 0x0, sizeof(bbl_ethernet_header_t)); *ethernet = eth; + eth->length = len; + /* Decode ethernet header */ header = (struct ether_header*)buf; BUMP_BUFFER(buf, len, sizeof(struct ether_header)); diff --git a/src/bbl_protocols.h b/src/bbl_protocols.h index 83385664..0894ad8d 100644 --- a/src/bbl_protocols.h +++ b/src/bbl_protocols.h @@ -358,8 +358,8 @@ typedef struct bbl_ethernet_header_ { uint8_t vlan_outer_priority; uint8_t vlan_inner_priority; void *next; // next header - uint32_t rx_sec; - uint32_t rx_nsec; + uint16_t length; + struct timespec timestamp; } bbl_ethernet_header_t; /* @@ -607,7 +607,7 @@ typedef struct bbl_bbl_ { uint32_t mc_group; uint64_t flow_id; uint64_t flow_seq; - uint64_t timestamp; + struct timespec timestamp; } bbl_bbl_t; typedef struct bbl_qmx_li_ { diff --git a/src/bbl_rx.c b/src/bbl_rx.c index 48e51492..20b774fe 100644 --- a/src/bbl_rx.c +++ b/src/bbl_rx.c @@ -9,6 +9,7 @@ #include "bbl.h" #include "bbl_session.h" +#include "bbl_stream.h" #include #include @@ -573,7 +574,7 @@ bbl_rx_dhcpv6(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *sessi bbl_udp_t *udp = (bbl_udp_t*)ipv6->next; bbl_dhcpv6_t *dhcpv6 = (bbl_dhcpv6_t*)udp->next; bbl_ctx_s *ctx = interface->ctx; - uint16_t tx_interval; + uint64_t tx_interval; if(dhcpv6->server_duid_len && dhcpv6->server_duid_len < DHCPV6_BUFFER) { memcpy(session->server_duid, dhcpv6->server_duid, dhcpv6->server_duid_len); @@ -599,16 +600,16 @@ bbl_rx_dhcpv6(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *sessi /* Start IPv6 PD Session Traffic */ if(bbl_add_session_packets_ipv6(ctx, session, true)) { if(ctx->config.session_traffic_ipv6pd_pps > 1) { - tx_interval = 1000 / ctx->config.session_traffic_ipv6pd_pps; + tx_interval = 1000000000 / ctx->config.session_traffic_ipv6pd_pps; if(tx_interval < ctx->config.tx_interval) { /* It is not possible to send faster than TX interval. */ tx_interval = ctx->config.tx_interval; } timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv6pd, "Session Traffic IPv6PD", - 0, tx_interval * MSEC, session, bbl_session_traffic_ipv6pd); + 0, tx_interval, session, bbl_session_traffic_ipv6pd); } else { timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv6pd, "Session Traffic IPv6PD", - 1, 0, session, bbl_session_traffic_ipv6pd); + 1, 0, session, bbl_session_traffic_ipv6pd); } } else { LOG(ERROR, "Traffic (ID: %u) failed to create IPv6 session traffic\n", session->session_id); @@ -637,7 +638,48 @@ bbl_rx_dhcpv6(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *sessi } void -bbl_rx_udp(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) { +bbl_rx_stream(bbl_interface_s *interface, bbl_ethernet_header_t *eth, bbl_bbl_t *bbl, uint8_t tos) { + + bbl_stream *stream; + void **search = NULL; + + struct timespec delay; + uint64_t delay_nsec; + + search = dict_search(interface->ctx->stream_flow_dict, &bbl->flow_id); + if(search) { + stream = *search; + stream->packets_rx++; + stream->rx_len = eth->length; + stream->rx_priority = tos; + stream->rx_outer_vlan_pbit = eth->vlan_outer_priority; + stream->rx_inner_vlan_pbit = eth->vlan_inner_priority; + + timespec_sub(&delay, ð->timestamp, &bbl->timestamp); + delay_nsec = delay.tv_sec * 1e9 + delay.tv_nsec; + if(delay_nsec > stream->max_delay_ns) { + stream->max_delay_ns = delay_nsec; + } + if(stream->min_delay_ns) { + if(delay_nsec > stream->min_delay_ns) { + stream->min_delay_ns = delay_nsec; + } + } else { + stream->min_delay_ns = delay_nsec; + } + if(!stream->rx_first_seq) { + stream->rx_first_seq = bbl->flow_seq; + } else { + if(stream->rx_last_seq +1 != bbl->flow_seq) { + stream->loss++; + } + } + stream->rx_last_seq = bbl->flow_seq; + } +} + +void +bbl_rx_udp(bbl_ethernet_header_t *eth, bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) { bbl_udp_t *udp = (bbl_udp_t*)ipv6->next; bbl_bbl_t *bbl = NULL; @@ -658,47 +700,31 @@ bbl_rx_udp(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) /* BBL receive handler */ if(bbl && bbl->type == BBL_TYPE_UNICAST_SESSION) { switch (bbl->sub_type) { - case BBL_SUB_TYPE_IPV4: - if(bbl->outer_vlan_id != session->vlan_key.outer_vlan_id || - bbl->inner_vlan_id != session->vlan_key.inner_vlan_id) { - interface->stats.session_ipv4_wrong_session++; - return; - } - interface->stats.session_ipv4_rx++; - session->stats.access_ipv4_rx++; - if(!session->access_ipv4_rx_first_seq) { - session->access_ipv4_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->access_ipv4_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv4_loss++; - session->stats.access_ipv4_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv4_rx_last_seq); - } - } - session->access_ipv4_rx_last_seq = bbl->flow_seq; - break; case BBL_SUB_TYPE_IPV6: if(bbl->outer_vlan_id != session->vlan_key.outer_vlan_id || bbl->inner_vlan_id != session->vlan_key.inner_vlan_id) { interface->stats.session_ipv6_wrong_session++; return; } - interface->stats.session_ipv6_rx++; - session->stats.access_ipv6_rx++; - if(!session->access_ipv6_rx_first_seq) { - session->access_ipv6_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->access_ipv6_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv6_loss++; - session->stats.access_ipv6_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv6_rx_last_seq); + if(bbl->flow_id == session->network_ipv4_tx_flow_id) { + /* Session traffic */ + interface->stats.session_ipv6_rx++; + session->stats.access_ipv6_rx++; + if(!session->access_ipv6_rx_first_seq) { + session->access_ipv6_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->access_ipv6_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv6_loss++; + session->stats.access_ipv6_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv6_rx_last_seq); + } } + session->access_ipv6_rx_last_seq = bbl->flow_seq; + } else { + bbl_rx_stream(interface, eth, bbl, ipv6->tos); } - session->access_ipv6_rx_last_seq = bbl->flow_seq; break; case BBL_SUB_TYPE_IPV6PD: if(bbl->outer_vlan_id != session->vlan_key.outer_vlan_id || @@ -706,20 +732,27 @@ bbl_rx_udp(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) interface->stats.session_ipv6pd_wrong_session++; return; } - interface->stats.session_ipv6pd_rx++; - session->stats.access_ipv6pd_rx++; - if(!session->access_ipv6pd_rx_first_seq) { - session->access_ipv6pd_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->access_ipv6pd_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv6pd_loss++; - session->stats.access_ipv6pd_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv6pd_rx_last_seq); + if(bbl->flow_id == session->network_ipv4_tx_flow_id) { + /* Session traffic */ + interface->stats.session_ipv6pd_rx++; + session->stats.access_ipv6pd_rx++; + if(!session->access_ipv6pd_rx_first_seq) { + session->access_ipv6pd_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->access_ipv6pd_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv6pd_loss++; + session->stats.access_ipv6pd_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv6pd_rx_last_seq); + } } + session->access_ipv6pd_rx_last_seq = bbl->flow_seq; + } else { + bbl_rx_stream(interface, eth, bbl, ipv6->tos); } - session->access_ipv6pd_rx_last_seq = bbl->flow_seq; + break; + default: break; } } @@ -730,7 +763,7 @@ bbl_rx_icmpv6(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *sessi bbl_icmpv6_t *icmpv6 = (bbl_icmpv6_t*)ipv6->next; bbl_ctx_s *ctx = interface->ctx; - uint16_t tx_interval; + uint64_t tx_interval; session->stats.icmpv6_rx++; if(icmpv6->type == IPV6_ICMPV6_ROUTER_ADVERTISEMENT) { @@ -753,16 +786,16 @@ bbl_rx_icmpv6(bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *sessi /* Start IPv6 Session Traffic */ if(bbl_add_session_packets_ipv6(ctx, session, false)) { if(ctx->config.session_traffic_ipv6_pps > 1) { - tx_interval = 1000 / ctx->config.session_traffic_ipv6_pps; + tx_interval = 1000000000 / ctx->config.session_traffic_ipv6_pps; if(tx_interval < ctx->config.tx_interval) { /* It is not possible to send faster than TX interval. */ tx_interval = ctx->config.tx_interval; } timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv6, "Session Traffic IPv6", - 0, tx_interval * MSEC, session, bbl_session_traffic_ipv6); + 0, tx_interval, session, bbl_session_traffic_ipv6); } else { timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv6, "Session Traffic IPv6", - 1, 0, session, bbl_session_traffic_ipv6); + 1, 0, session, bbl_session_traffic_ipv6); } } else { LOG(ERROR, "Traffic (ID: %u) failed to create IPv6 session traffic\n", session->session_id); @@ -899,21 +932,25 @@ bbl_rx_ipv4(bbl_ethernet_header_t *eth, bbl_ipv4_t *ipv4, bbl_interface_s *inter interface->stats.session_ipv4_wrong_session++; return; } - interface->stats.session_ipv4_rx++; - session->stats.access_ipv4_rx++; - if(!session->access_ipv4_rx_first_seq) { - session->access_ipv4_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->access_ipv4_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv4_loss++; - session->stats.access_ipv4_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv4_rx_last_seq); + if(bbl->flow_id == session->network_ipv4_tx_flow_id) { + /* Session traffic */ + interface->stats.session_ipv4_rx++; + session->stats.access_ipv4_rx++; + if(!session->access_ipv4_rx_first_seq) { + session->access_ipv4_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->access_ipv4_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv4_loss++; + session->stats.access_ipv4_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->access_ipv4_rx_last_seq); + } } + session->access_ipv4_rx_last_seq = bbl->flow_seq; + } else { + bbl_rx_stream(interface, eth, bbl, ipv4->tos); } - session->access_ipv4_rx_last_seq = bbl->flow_seq; - } else if(bbl->type == BBL_TYPE_MULTICAST) { /* Multicast receive handler */ for(i=0; i < IGMP_MAX_GROUPS; i++) { @@ -924,8 +961,8 @@ bbl_rx_ipv4(bbl_ethernet_header_t *eth, bbl_ipv4_t *ipv4, bbl_interface_s *inter session->stats.mc_rx++; group->packets++; if(!group->first_mc_rx_time.tv_sec) { - group->first_mc_rx_time.tv_sec = eth->rx_sec; - group->first_mc_rx_time.tv_nsec = eth->rx_nsec; + group->first_mc_rx_time.tv_sec = eth->timestamp.tv_sec; + group->first_mc_rx_time.tv_nsec = eth->timestamp.tv_nsec; } else if(bbl->flow_seq > session->mc_rx_last_seq + 1) { interface->stats.mc_loss++; session->stats.mc_loss++; @@ -938,8 +975,8 @@ bbl_rx_ipv4(bbl_ethernet_header_t *eth, bbl_ipv4_t *ipv4, bbl_interface_s *inter interface->stats.mc_rx++; session->stats.mc_rx++; group->packets++; - group->last_mc_rx_time.tv_sec = eth->rx_sec; - group->last_mc_rx_time.tv_nsec = eth->rx_nsec; + group->last_mc_rx_time.tv_sec = eth->timestamp.tv_sec; + group->last_mc_rx_time.tv_nsec = eth->timestamp.tv_nsec; if(session->zapping_joined_group && session->zapping_leaved_group == group) { if(session->zapping_joined_group->first_mc_rx_time.tv_sec) { @@ -961,15 +998,15 @@ bbl_rx_ipv4(bbl_ethernet_header_t *eth, bbl_ipv4_t *ipv4, bbl_interface_s *inter session->stats.mc_rx++; group->packets++; if(!group->first_mc_rx_time.tv_sec) { - group->first_mc_rx_time.tv_sec = eth->rx_sec; - group->first_mc_rx_time.tv_nsec = eth->rx_nsec; + group->first_mc_rx_time.tv_sec = eth->timestamp.tv_sec; + group->first_mc_rx_time.tv_nsec = eth->timestamp.tv_nsec; } } else { interface->stats.mc_rx++; session->stats.mc_rx++; group->packets++; - group->last_mc_rx_time.tv_sec = eth->rx_sec; - group->last_mc_rx_time.tv_nsec = eth->rx_nsec; + group->last_mc_rx_time.tv_sec = eth->timestamp.tv_sec; + group->last_mc_rx_time.tv_nsec = eth->timestamp.tv_nsec; if(session->zapping_joined_group && session->zapping_leaved_group == group) { if(session->zapping_joined_group->first_mc_rx_time.tv_sec) { @@ -983,14 +1020,14 @@ bbl_rx_ipv4(bbl_ethernet_header_t *eth, bbl_ipv4_t *ipv4, bbl_interface_s *inter } void -bbl_rx_ipv6(bbl_ethernet_header_t *eth __attribute__((unused)), bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) { +bbl_rx_ipv6(bbl_ethernet_header_t *eth, bbl_ipv6_t *ipv6, bbl_interface_s *interface, bbl_session_s *session) { switch(ipv6->protocol) { case IPV6_NEXT_HEADER_ICMPV6: bbl_rx_icmpv6(ipv6, interface, session); interface->stats.icmpv6_rx++; break; case IPV6_NEXT_HEADER_UDP: - bbl_rx_udp(ipv6, interface, session); + bbl_rx_udp(eth, ipv6, interface, session); break; default: break; @@ -1125,7 +1162,7 @@ bbl_rx_established(bbl_ethernet_header_t *eth, bbl_interface_s *interface, bbl_s bool ipcp = false; bool ip6cp = false; - uint16_t tx_interval; + uint64_t tx_interval; if(ctx->config.ipcp_enable == false || session->ipcp_state == BBL_PPP_OPENED) ipcp = true; if(ctx->config.ip6cp_enable == false || session->ip6cp_state == BBL_PPP_OPENED) ip6cp = true; @@ -1133,8 +1170,8 @@ bbl_rx_established(bbl_ethernet_header_t *eth, bbl_interface_s *interface, bbl_s if(ipcp && ip6cp) { if(session->session_state != BBL_ESTABLISHED) { if(ctx->sessions_established_max < ctx->sessions) { - ctx->stats.last_session_established.tv_sec = eth->rx_sec; - ctx->stats.last_session_established.tv_nsec = eth->rx_nsec; + ctx->stats.last_session_established.tv_sec = eth->timestamp.tv_sec; + ctx->stats.last_session_established.tv_nsec = eth->timestamp.tv_nsec; bbl_session_update_state(ctx, session, BBL_ESTABLISHED); if(ctx->sessions_established == ctx->sessions) { LOG(NORMAL, "ALL SESSIONS ESTABLISHED\n"); @@ -1159,16 +1196,16 @@ bbl_rx_established(bbl_ethernet_header_t *eth, bbl_interface_s *interface, bbl_s /* Start IPv4 Session Traffic */ if(bbl_add_session_packets_ipv4(ctx, session)) { if(ctx->config.session_traffic_ipv4_pps > 1) { - tx_interval = 1000 / ctx->config.session_traffic_ipv4_pps; + tx_interval = 1000000000 / ctx->config.session_traffic_ipv4_pps; if(tx_interval < ctx->config.tx_interval) { /* It is not possible to send faster than TX interval. */ tx_interval = ctx->config.tx_interval; } timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv4, "Session Traffic IPv4", - 0, tx_interval * MSEC, session, bbl_session_traffic_ipv4); + 0, tx_interval, session, bbl_session_traffic_ipv4); } else { timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv4, "Session Traffic IPv4", - 1, 0, session, bbl_session_traffic_ipv4); + 1, 0, session, bbl_session_traffic_ipv4); } } else { LOG(ERROR, "Traffic (ID: %u) failed to create IPv4 session traffic\n", session->session_id); @@ -1182,12 +1219,12 @@ void bbl_rx_established_ipoe(bbl_ethernet_header_t *eth, bbl_interface_s *interface, bbl_session_s *session) { bbl_ctx_s *ctx = interface->ctx; - uint16_t tx_interval; + uint64_t tx_interval; if(session->session_state != BBL_ESTABLISHED) { if(ctx->sessions_established_max < ctx->sessions) { - ctx->stats.last_session_established.tv_sec = eth->rx_sec; - ctx->stats.last_session_established.tv_nsec = eth->rx_nsec; + ctx->stats.last_session_established.tv_sec = eth->timestamp.tv_sec; + ctx->stats.last_session_established.tv_nsec = eth->timestamp.tv_nsec; bbl_session_update_state(ctx, session, BBL_ESTABLISHED); if(ctx->sessions_established == ctx->sessions) { LOG(NORMAL, "ALL SESSIONS ESTABLISHED\n"); @@ -1204,16 +1241,16 @@ bbl_rx_established_ipoe(bbl_ethernet_header_t *eth, bbl_interface_s *interface, /* Start IPv4 Session Traffic */ if(bbl_add_session_packets_ipv4(ctx, session)) { if(ctx->config.session_traffic_ipv4_pps > 1) { - tx_interval = 1000 / ctx->config.session_traffic_ipv4_pps; + tx_interval = 1000000000 / ctx->config.session_traffic_ipv4_pps; if(tx_interval < ctx->config.tx_interval) { /* It is not possible to send faster than TX interval. */ tx_interval = ctx->config.tx_interval; } timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv4, "Session Traffic IPv4", - 0, tx_interval * MSEC, session, bbl_session_traffic_ipv4); + 0, tx_interval, session, bbl_session_traffic_ipv4); } else { timer_add_periodic(&ctx->timer_root, &session->timer_session_traffic_ipv4, "Session Traffic IPv4", - 1, 0, session, bbl_session_traffic_ipv4); + 1, 0, session, bbl_session_traffic_ipv4); } } else { LOG(ERROR, "Traffic (ID: %u) failed to create IPv4 session traffic\n", session->session_id); @@ -1835,9 +1872,9 @@ bbl_rx_handler_network(bbl_ethernet_header_t *eth, bbl_interface_s *interface) { bbl_ctx_s *ctx; - bbl_ipv4_t *ipv4; - bbl_ipv6_t *ipv6; - bbl_udp_t *udp; + bbl_ipv4_t *ipv4 = NULL; + bbl_ipv6_t *ipv6 = NULL; + bbl_udp_t *udp = NULL; bbl_bbl_t *bbl = NULL; bbl_session_s *session; @@ -1888,52 +1925,64 @@ bbl_rx_handler_network(bbl_ethernet_header_t *eth, bbl_interface_s *interface) { if(session) { switch (bbl->sub_type) { case BBL_SUB_TYPE_IPV4: - interface->stats.session_ipv4_rx++; - session->stats.network_ipv4_rx++; - if(!session->network_ipv4_rx_first_seq) { - session->network_ipv4_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->network_ipv4_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv4_loss++; - session->stats.network_ipv4_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv4_rx_last_seq); + if(session->access_ipv4_tx_flow_id == bbl->flow_id) { + interface->stats.session_ipv4_rx++; + session->stats.network_ipv4_rx++; + if(!session->network_ipv4_rx_first_seq) { + session->network_ipv4_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->network_ipv4_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv4_loss++; + session->stats.network_ipv4_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv4_rx_last_seq); + } } + session->network_ipv4_rx_last_seq = bbl->flow_seq; + } else { + if(ipv4) bbl_rx_stream(interface, eth, bbl, ipv4->tos); } - session->network_ipv4_rx_last_seq = bbl->flow_seq; break; case BBL_SUB_TYPE_IPV6: - interface->stats.session_ipv6_rx++; - session->stats.network_ipv6_rx++; - if(!session->network_ipv6_rx_first_seq) { - session->network_ipv6_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->network_ipv6_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv6_loss++; - session->stats.network_ipv6_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv6_rx_last_seq); + if(session->access_ipv6_tx_flow_id == bbl->flow_id) { + interface->stats.session_ipv6_rx++; + session->stats.network_ipv6_rx++; + if(!session->network_ipv6_rx_first_seq) { + session->network_ipv6_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->network_ipv6_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv6_loss++; + session->stats.network_ipv6_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv6_rx_last_seq); + } } + session->network_ipv6_rx_last_seq = bbl->flow_seq; + } else { + if(ipv6) bbl_rx_stream(interface, eth, bbl, ipv6->tos); } - session->network_ipv6_rx_last_seq = bbl->flow_seq; break; case BBL_SUB_TYPE_IPV6PD: - interface->stats.session_ipv6pd_rx++; - session->stats.network_ipv6pd_rx++; - if(!session->network_ipv6pd_rx_first_seq) { - session->network_ipv6pd_rx_first_seq = bbl->flow_seq; - interface->ctx->stats.session_traffic_flows_verified++; - } else { - if(session->network_ipv6pd_rx_last_seq +1 != bbl->flow_seq) { - interface->stats.session_ipv6pd_loss++; - session->stats.network_ipv6pd_loss++; - LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", - session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv6pd_rx_last_seq); + if(session->access_ipv6pd_tx_flow_id == bbl->flow_id) { + interface->stats.session_ipv6pd_rx++; + session->stats.network_ipv6pd_rx++; + if(!session->network_ipv6pd_rx_first_seq) { + session->network_ipv6pd_rx_first_seq = bbl->flow_seq; + interface->ctx->stats.session_traffic_flows_verified++; + } else { + if(session->network_ipv6pd_rx_last_seq +1 != bbl->flow_seq) { + interface->stats.session_ipv6pd_loss++; + session->stats.network_ipv6pd_loss++; + LOG(LOSS, "LOSS (ID: %u) flow: %lu seq: %lu last: %lu\n", + session->session_id, bbl->flow_id, bbl->flow_seq, session->network_ipv6pd_rx_last_seq); + } } + session->network_ipv6pd_rx_last_seq = bbl->flow_seq; + } else { + if(ipv6) bbl_rx_stream(interface, eth, bbl, ipv6->tos); } - session->network_ipv6pd_rx_last_seq = bbl->flow_seq; break; default: break; diff --git a/src/bbl_session.c b/src/bbl_session.c index e4f746a5..56542c1d 100644 --- a/src/bbl_session.c +++ b/src/bbl_session.c @@ -10,6 +10,7 @@ #include "bbl.h" #include "bbl_session.h" +#include "bbl_stream.h" extern volatile bool g_teardown; @@ -446,6 +447,12 @@ bbl_sessions_init(bbl_ctx_s *ctx) *result.datum_ptr = session; } } + if(access_config->stream_group_id) { + if(!bbl_stream_add(ctx, access_config, session)) { + LOG(ERROR, "Failed to create session traffic stream!\n"); + return false; + } + } LOG(DEBUG, "Session %u created (%s.%u:%u)\n", i, access_config->interface, access_config->access_outer_vlan, access_config->access_inner_vlan); i++; diff --git a/src/bbl_stats.h b/src/bbl_stats.h index 011d27ef..af2951aa 100644 --- a/src/bbl_stats.h +++ b/src/bbl_stats.h @@ -43,6 +43,7 @@ typedef struct bbl_stats_ { uint32_t sessions_network_ipv6pd_rx; } bbl_stats_t; +void bbl_compute_avg_rate (bbl_rate_s *rate, uint64_t current_value); void bbl_stats_update_cps (bbl_ctx_s *ctx); void bbl_stats_generate(bbl_ctx_s *ctx, bbl_stats_t *stats); void bbl_stats_stdout(bbl_ctx_s *ctx, bbl_stats_t *stats); diff --git a/src/bbl_stream.c b/src/bbl_stream.c new file mode 100644 index 00000000..10181e25 --- /dev/null +++ b/src/bbl_stream.c @@ -0,0 +1,638 @@ +/* + * BNG Blaster (BBL) - Streams + * + * Christian Giese, March 2021 + * + * Copyright (C) 2020-2021, RtBrick, Inc. + */ + +#include "bbl.h" +#include "bbl_stream.h" +#include "bbl_stats.h" +#include "bbl_io_packet_mmap.h" +#include "bbl_io_raw.h" +#ifdef BNGBLASTER_NETMAP + #include "bbl_io_netmap.h" +#endif + + +bool +bbl_stream_can_send(bbl_stream *stream) { + bbl_session_s *session = stream->session; + + if(session->session_state == BBL_ESTABLISHED) { + if(session->access_type == ACCESS_TYPE_PPPOE) { + switch (stream->config->type) { + case STREAM_L2TP: + case STREAM_IPV4: + if(session->ipcp_state == BBL_PPP_OPENED) { + return true; + } + break; + case STREAM_IPV6: + if(session->ip6cp_state == BBL_PPP_OPENED && session->icmpv6_ra_received) { + return true; + } + break; + case STREAM_IPV6PD: + if(session->ip6cp_state == BBL_PPP_OPENED && session->dhcpv6_received) { + return true; + } + break; + default: + break; + } + } + } + /* Free of packet if not ready to send */ + if(stream->buf) { + free(stream->buf); + stream->buf = NULL; + stream->tx_len = 0; + } + return false; +} + +bool +bbl_stream_build_access_pppoe_packet(bbl_stream *stream) { + + bbl_ctx_s *ctx = stream->interface->ctx; + bbl_session_s *session = stream->session; + bbl_stream_config *config = stream->config; + + uint16_t buf_len; + + bbl_ethernet_header_t eth = {0}; + bbl_pppoe_session_t pppoe = {0}; + bbl_ipv4_t ipv4 = {0}; + bbl_ipv6_t ipv6 = {0}; + bbl_udp_t udp = {0}; + bbl_bbl_t bbl = {0}; + + eth.dst = session->server_mac; + eth.src = session->client_mac; + eth.vlan_outer = session->vlan_key.outer_vlan_id; + eth.vlan_outer_priority = config->vlan_priority; + eth.vlan_inner = session->vlan_key.inner_vlan_id; + eth.vlan_inner_priority = config->vlan_priority; + eth.vlan_three = session->access_third_vlan; + eth.type = ETH_TYPE_PPPOE_SESSION; + eth.next = &pppoe; + pppoe.session_id = session->pppoe_session_id; + udp.src = BBL_UDP_PORT; + udp.dst = BBL_UDP_PORT; + udp.protocol = UDP_PROTOCOL_BBL; + udp.next = &bbl; + bbl.type = BBL_TYPE_UNICAST_SESSION; + bbl.session_id = session->session_id; + bbl.ifindex = session->interface->ifindex; + bbl.outer_vlan_id = session->vlan_key.outer_vlan_id; + bbl.inner_vlan_id = session->vlan_key.inner_vlan_id; + bbl.flow_id = stream->flow_id; + bbl.direction = BBL_DIRECTION_UP; + + switch (stream->config->type) { + case STREAM_L2TP: + case STREAM_IPV4: + pppoe.protocol = PROTOCOL_IPV4; + pppoe.next = &ipv4; + ipv4.dst = ctx->op.network_if->ip; + ipv4.src = session->ip_address; + ipv4.ttl = 64; + ipv4.tos = config->priority; + ipv4.protocol = PROTOCOL_IPV4_UDP; + ipv4.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV4; + if (config->length > 76) { + bbl.padding = config->length - 76; + } + break; + case STREAM_IPV6: + pppoe.protocol = PROTOCOL_IPV6; + pppoe.next = &ipv6; + ipv6.dst = ctx->op.network_if->ip6.address; + ipv6.src = session->ipv6_address; + ipv6.ttl = 64; + ipv4.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + case STREAM_IPV6PD: + pppoe.protocol = PROTOCOL_IPV6; + pppoe.next = &ipv6; + ipv6.dst = ctx->op.network_if->ip6.address; + ipv6.src = session->delegated_ipv6_address; + ipv6.ttl = 64; + ipv4.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + default: + return false; + } + + buf_len = config->length + 64; + if(buf_len < 256) buf_len = 256; + stream->buf = malloc(buf_len); + if(encode_ethernet(stream->buf, &stream->tx_len, ð) != PROTOCOL_SUCCESS) { + free(stream->buf); + stream->buf = NULL; + stream->tx_len = 0; + return false; + } + return true; +} + +bool +bbl_stream_build_access_ipoe_packet(bbl_stream *stream) { + + bbl_ctx_s *ctx = stream->interface->ctx; + bbl_session_s *session = stream->session; + bbl_stream_config *config = stream->config; + + uint16_t buf_len; + + bbl_ethernet_header_t eth = {0}; + bbl_ipv4_t ipv4 = {0}; + bbl_ipv6_t ipv6 = {0}; + bbl_udp_t udp = {0}; + bbl_bbl_t bbl = {0}; + + eth.dst = session->server_mac; + eth.src = session->client_mac; + eth.vlan_outer = session->vlan_key.outer_vlan_id; + eth.vlan_outer_priority = config->vlan_priority; + eth.vlan_inner = session->vlan_key.inner_vlan_id; + eth.vlan_inner_priority = config->vlan_priority; + eth.vlan_three = session->access_third_vlan; + + udp.src = BBL_UDP_PORT; + udp.dst = BBL_UDP_PORT; + udp.protocol = UDP_PROTOCOL_BBL; + udp.next = &bbl; + bbl.type = BBL_TYPE_UNICAST_SESSION; + bbl.session_id = session->session_id; + bbl.ifindex = session->interface->ifindex; + bbl.outer_vlan_id = session->vlan_key.outer_vlan_id; + bbl.inner_vlan_id = session->vlan_key.inner_vlan_id; + bbl.flow_id = stream->flow_id; + bbl.direction = BBL_DIRECTION_UP; + + switch (stream->config->type) { + case STREAM_IPV4: + eth.type = ETH_TYPE_IPV4; + eth.next = &ipv4; + ipv4.dst = ctx->op.network_if->ip; + ipv4.src = session->ip_address; + ipv4.ttl = 64; + ipv4.tos = config->priority; + ipv4.protocol = PROTOCOL_IPV4_UDP; + ipv4.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV4; + if (config->length > 76) { + bbl.padding = config->length - 76; + } + break; + case STREAM_IPV6: + eth.type = ETH_TYPE_IPV6; + eth.next = &ipv6; + ipv6.dst = ctx->op.network_if->ip6.address; + ipv6.src = session->ipv6_address; + ipv6.ttl = 64; + ipv6.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + case STREAM_IPV6PD: + eth.type = ETH_TYPE_IPV6; + eth.next = &ipv6; + ipv6.dst = ctx->op.network_if->ip6.address; + ipv6.src = session->delegated_ipv6_address; + ipv6.ttl = 64; + ipv6.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + default: + return false; + } + + buf_len = config->length + 64; + if(buf_len < 256) buf_len = 256; + stream->buf = malloc(buf_len); + if(encode_ethernet(stream->buf, &stream->tx_len, ð) != PROTOCOL_SUCCESS) { + free(stream->buf); + stream->buf = NULL; + stream->tx_len = 0; + return false; + } + return true; +} + +bool +bbl_stream_build_network_packet(bbl_stream *stream) { + + bbl_ctx_s *ctx = stream->interface->ctx; + bbl_session_s *session = stream->session; + bbl_stream_config *config = stream->config; + + uint16_t buf_len; + + bbl_ethernet_header_t eth = {0}; + bbl_ipv4_t ipv4 = {0}; + bbl_ipv6_t ipv6 = {0}; + bbl_udp_t udp = {0}; + bbl_bbl_t bbl = {0}; + + eth.dst = ctx->op.network_if->gateway_mac; + eth.src = ctx->op.network_if->mac; + eth.vlan_outer = ctx->config.network_vlan; + eth.vlan_outer_priority = config->vlan_priority; + eth.vlan_inner = 0; + + udp.src = BBL_UDP_PORT; + udp.dst = BBL_UDP_PORT; + udp.protocol = UDP_PROTOCOL_BBL; + udp.next = &bbl; + bbl.type = BBL_TYPE_UNICAST_SESSION; + bbl.session_id = session->session_id; + bbl.ifindex = session->interface->ifindex; + bbl.outer_vlan_id = session->vlan_key.outer_vlan_id; + bbl.inner_vlan_id = session->vlan_key.inner_vlan_id; + bbl.flow_id = stream->flow_id; + bbl.direction = BBL_DIRECTION_DOWN; + + switch (stream->config->type) { + case STREAM_IPV4: + eth.type = ETH_TYPE_IPV4; + eth.next = &ipv4; + ipv4.dst = session->ip_address; + ipv4.src = ctx->op.network_if->ip; + ipv4.ttl = 64; + ipv4.tos = config->priority; + ipv4.protocol = PROTOCOL_IPV4_UDP; + ipv4.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV4; + if (config->length > 76) { + bbl.padding = config->length - 76; + } + break; + case STREAM_IPV6: + eth.type = ETH_TYPE_IPV6; + eth.next = &ipv6; + ipv6.dst = session->ipv6_address; + ipv6.src = ctx->op.network_if->ip6.address; + ipv6.ttl = 64; + ipv6.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + case STREAM_IPV6PD: + eth.type = ETH_TYPE_IPV6; + eth.next = &ipv6; + ipv6.dst = session->delegated_ipv6_address; + ipv6.src = ctx->op.network_if->ip6.address; + ipv6.ttl = 64; + ipv6.tos = config->priority; + ipv6.protocol = IPV6_NEXT_HEADER_UDP; + ipv6.next = &udp; + bbl.sub_type = BBL_SUB_TYPE_IPV6; + if (config->length > 96) { + bbl.padding = config->length - 96; + } + break; + default: + return false; + } + + buf_len = config->length + 64; + if(buf_len < 256) buf_len = 256; + stream->buf = malloc(buf_len); + if(encode_ethernet(stream->buf, &stream->tx_len, ð) != PROTOCOL_SUCCESS) { + free(stream->buf); + stream->buf = NULL; + stream->tx_len = 0; + return false; + } + return true; +} + +bool +bbl_stream_build_l2tp_packet(bbl_stream *stream) { + + bbl_ctx_s *ctx = stream->interface->ctx; + bbl_session_s *session = stream->session; + bbl_stream_config *config = stream->config; + + bbl_l2tp_session_t *l2tp_session = stream->l2tp_session; + bbl_l2tp_tunnel_t *l2tp_tunnel = l2tp_session->tunnel; + + uint16_t buf_len; + + bbl_ethernet_header_t eth = {0}; + bbl_ipv4_t l2tp_ipv4 = {0}; + bbl_udp_t l2tp_udp = {0}; + bbl_l2tp_t l2tp = {0}; + bbl_ipv4_t ipv4 = {0}; + bbl_udp_t udp = {0}; + bbl_bbl_t bbl = {0}; + + eth.dst = ctx->op.network_if->gateway_mac; + eth.src = ctx->op.network_if->mac; + eth.vlan_outer = ctx->config.network_vlan; + eth.vlan_inner = 0; + eth.type = ETH_TYPE_IPV4; + eth.next = &l2tp_ipv4; + l2tp_ipv4.dst = l2tp_tunnel->peer_ip; + l2tp_ipv4.src = l2tp_tunnel->server->ip; + l2tp_ipv4.ttl = 64; + l2tp_ipv4.tos = config->priority; + l2tp_ipv4.protocol = PROTOCOL_IPV4_UDP; + l2tp_ipv4.next = &l2tp_udp; + l2tp_udp.src = L2TP_UDP_PORT; + l2tp_udp.dst = L2TP_UDP_PORT; + l2tp_udp.protocol = UDP_PROTOCOL_L2TP; + l2tp_udp.next = &l2tp; + l2tp.type = L2TP_MESSAGE_DATA; + l2tp.tunnel_id = l2tp_tunnel->peer_tunnel_id; + l2tp.session_id = l2tp_session->peer_session_id; + l2tp.protocol = PROTOCOL_IPV4; + l2tp.with_length = l2tp_tunnel->server->data_lenght; + l2tp.with_offset = l2tp_tunnel->server->data_offset; + l2tp.next = &ipv4; + ipv4.dst = session->ip_address; + ipv4.src = l2tp_tunnel->server->ip; + ipv4.ttl = 64; + ipv4.protocol = PROTOCOL_IPV4_UDP; + ipv4.next = &udp; + udp.src = BBL_UDP_PORT; + udp.dst = BBL_UDP_PORT; + udp.protocol = UDP_PROTOCOL_BBL; + udp.next = &bbl; + bbl.type = BBL_TYPE_UNICAST_SESSION; + bbl.session_id = session->session_id; + bbl.ifindex = session->interface->ifindex; + bbl.outer_vlan_id = session->vlan_key.outer_vlan_id; + bbl.inner_vlan_id = session->vlan_key.inner_vlan_id; + bbl.flow_id = stream->flow_id; + bbl.direction = BBL_DIRECTION_DOWN; + bbl.sub_type = BBL_SUB_TYPE_IPV4; + if (config->length > 76) { + bbl.padding = config->length - 76; + } + + buf_len = config->length + 128; + if(buf_len < 256) buf_len = 256; + stream->buf = malloc(buf_len); + if(encode_ethernet(stream->buf, &stream->tx_len, ð) != PROTOCOL_SUCCESS) { + free(stream->buf); + stream->buf = NULL; + stream->tx_len = 0; + return false; + } + return true; +} + +bool +bbl_stream_build_packet(bbl_stream *stream) { + + if(stream->session->access_type == ACCESS_TYPE_PPPOE) { + switch (stream->config->type) { + case STREAM_IPV4: + case STREAM_IPV6: + case STREAM_IPV6PD: + if(stream->direction == STREAM_DIRECTION_UP) { + return bbl_stream_build_access_pppoe_packet(stream); + } else { + return bbl_stream_build_network_packet(stream); + } + break; + case STREAM_L2TP: + if(stream->direction == STREAM_DIRECTION_UP) { + return bbl_stream_build_access_pppoe_packet(stream); + } else { + if(stream->l2tp_session) { + return bbl_stream_build_l2tp_packet(stream); + } + } + default: + break; + } + } else if (stream->session->access_type == ACCESS_TYPE_IPOE) { + if(stream->direction == STREAM_DIRECTION_UP) { + return bbl_stream_build_access_ipoe_packet(stream); + } else { + return bbl_stream_build_network_packet(stream); + } + } + return false; +} + +void +bbl_stream_tx_job (timer_s *timer) { + + bbl_stream *stream = timer->data; + bbl_session_s *session = stream->session; + bbl_interface_s *interface = stream->interface; + + struct timespec send_windwow; + struct timespec now; + + double d; + uint64_t packets = 1; + uint64_t packets_expected; + + if(!bbl_stream_can_send(stream)) { + stream->send_window_packets = 0; + return; + } + if(!stream->buf) { + if(!bbl_stream_build_packet(stream)) { + LOG(ERROR, "Failed to build packet for stream %s session-id %u\n", + stream->config->name, session->session_id); + return; + } + } + + clock_gettime(CLOCK_MONOTONIC, &now); + if(stream->send_window_packets == 0) { + /* Open new send window */ + stream->send_window_start.tv_sec = now.tv_sec; + stream->send_window_start.tv_nsec = now.tv_nsec; + } else { + timespec_sub(&send_windwow, &now, &stream->send_window_start); + packets_expected = send_windwow.tv_sec * stream->config->pps; + d = (send_windwow.tv_nsec / 1000000000.0); + packets_expected += d * stream->config->pps; + + if(packets_expected > stream->send_window_packets) { + packets = packets_expected - stream->send_window_packets; + } + if(packets > (interface->ctx->config.io_slots >> 1)) { + packets = (interface->ctx->config.io_slots >> 1); + } + } + + /* Update BBL header fields */ + clock_gettime(CLOCK_REALTIME, &interface->tx_timestamp); + *(uint32_t*)(stream->buf + (stream->tx_len - 8)) = interface->tx_timestamp.tv_sec; + *(uint32_t*)(stream->buf + (stream->tx_len - 4)) = interface->tx_timestamp.tv_nsec; + while(packets) { + *(uint64_t*)(stream->buf + (stream->tx_len - 16)) = stream->flow_seq; + /* Send packet ... */ + switch (interface->io_mode) { + case IO_MODE_PACKET_MMAP: + if(!bbl_io_packet_mmap_send(interface, stream->buf, stream->tx_len)) { + return; + } + break; + case IO_MODE_RAW: + if(!bbl_io_raw_send(interface, stream->buf, stream->tx_len)) { + return; + } + break; + #ifdef BNGBLASTER_NETMAP + case IO_MODE_NETMAP: + if(!bbl_io_netmap_send(interface, stream->buf, stream->tx_len)) { + return; + } + break; + #endif + default: + return; + } + stream->send_window_packets++; + stream->packets_tx++; + stream->flow_seq++; + packets--; + } +} + +void +bbl_stream_rate_job (timer_s *timer) { + bbl_stream *stream = timer->data; + bbl_compute_avg_rate(&stream->rate_packets_tx, stream->packets_tx); + bbl_compute_avg_rate(&stream->rate_packets_rx, stream->packets_rx); +} + +bool +bbl_stream_add(bbl_ctx_s *ctx, bbl_access_config_s *access_config, bbl_session_s *session) { + + bbl_stream_config *config; + bbl_stream *stream; + bbl_stream *session_stream; + + dict_insert_result result; + + time_t timer_sec = 0; + long timer_nsec = 0; + + config = ctx->config.stream_config; + + while(config) { + if(config->stream_group_id == access_config->stream_group_id) { + + if(!ctx->op.network_if) { + LOG(ERROR, "Failed to add stream because of missing network interface\n"); + return false; + } + + if(config->pps == 1) { + timer_sec = 1; + } else { + timer_nsec = 1000000000 / config->pps; + } + + if(config->direction & STREAM_DIRECTION_UP) { + stream = calloc(1, sizeof(bbl_stream)); + stream->flow_id = ctx->flow_id++; + stream->flow_seq = 1; + stream->config = config; + stream->direction = STREAM_DIRECTION_UP; + stream->interface = session->interface; + stream->session = session; + stream->tx_interval = timer_sec * 1e9 + timer_nsec; + result = dict_insert(ctx->stream_flow_dict, &stream->flow_id); + if (!result.inserted) { + LOG(ERROR, "Failed to insert stream %s\n", config->name); + free(stream); + return false; + } + *result.datum_ptr = stream; + if(session->stream) { + session_stream = session->stream; + while(session_stream->next) { + session_stream = session_stream->next; + } + session_stream->next = stream; + } else { + session->stream = stream; + } + if(stream->tx_interval < ctx->config.tx_interval) { + timer_sec = 0; + timer_nsec = ctx->config.tx_interval; + } + timer_add_periodic(&ctx->timer_root, &stream->timer, config->name, timer_sec, timer_nsec, stream, bbl_stream_tx_job); + timer_add_periodic(&ctx->timer_root, &stream->timer_rate, "Rate Computation", 1, 0, stream, bbl_stream_rate_job); + LOG(DEBUG, "Traffic stream %s added in upstream with timer %lu sec %lu nsec\n", config->name, timer_sec, timer_nsec); + } + if(config->direction & STREAM_DIRECTION_DOWN) { + stream = calloc(1, sizeof(bbl_stream)); + stream->flow_id = ctx->flow_id++; + stream->flow_seq = 1; + stream->config = config; + stream->direction = STREAM_DIRECTION_DOWN; + stream->interface = ctx->op.network_if; + stream->session = session; + stream->tx_interval = timer_sec * 1e9 + timer_nsec; + result = dict_insert(ctx->stream_flow_dict, &stream->flow_id); + if (!result.inserted) { + LOG(ERROR, "Failed to insert stream %s\n", config->name); + free(stream); + return false; + } + *result.datum_ptr = stream; + if(session->stream) { + session_stream = session->stream; + while(session_stream->next) { + session_stream = session_stream->next; + } + session_stream->next = stream; + } else { + session->stream = stream; + } + if(stream->tx_interval < ctx->config.tx_interval) { + timer_sec = 0; + timer_nsec = ctx->config.tx_interval; + } + timer_add_periodic(&ctx->timer_root, &stream->timer, config->name, timer_sec, timer_nsec, stream, bbl_stream_tx_job); + timer_add_periodic(&ctx->timer_root, &stream->timer_rate, "Rate Computation", 1, 0, stream, bbl_stream_rate_job); + LOG(DEBUG, "Traffic stream %s added in downstream with timer %lu sec %lu nsec\n", config->name, timer_sec, timer_nsec); + } + timer_smear_bucket(&ctx->timer_root, timer_sec, timer_nsec); + } + config = config->next; + } + + return true; +} \ No newline at end of file diff --git a/src/bbl_stream.h b/src/bbl_stream.h new file mode 100644 index 00000000..5d2b47f2 --- /dev/null +++ b/src/bbl_stream.h @@ -0,0 +1,92 @@ +/* + * BNG Blaster (BBL) - Streams + * + * Christian Giese, Match 2021 + * + * Copyright (C) 2020-2021, RtBrick, Inc. + */ + +#ifndef __BBL_STREAM_H__ +#define __BBL_STREAM_H__ + +typedef enum { + STREAM_IPV4, /* From/to framed IPv4 address */ + STREAM_IPV6, /* From/to framed IPv6 address */ + STREAM_IPV6PD, /* From/to delegated IPv6 address */ + STREAM_L2TP, /* L2TP downstream (LNS to client) */ +} __attribute__ ((__packed__)) bbl_stream_type_t; + +typedef enum { + STREAM_DIRECTION_UP = 1, + STREAM_DIRECTION_DOWN = 2, + STREAM_DIRECTION_BOTH = 3 +} __attribute__ ((__packed__)) bbl_stream_direction_t; + +typedef struct bbl_stream_config_ +{ + char *name; + uint16_t stream_group_id; + + bbl_stream_type_t type; + bbl_stream_direction_t direction; + + uint32_t pps; + uint16_t length; + uint8_t priority; /* IPv4 TOS or IPv6 TC */ + uint8_t vlan_priority; + + uint32_t ipv4_source_address; /* overwrite default IPv4 source address */ + ipv6addr_t ipv6_source_address; /* overwrite default IPv6 source address */ + + void *next; /* next bbl_stream_config */ +} bbl_stream_config; + +typedef struct bbl_stream_ +{ + uint64_t flow_id; + uint64_t flow_seq; + + struct timer_ *timer; + struct timer_ *timer_rate; + + bbl_stream_config *config; + bbl_stream_direction_t direction; + + bbl_interface_s *interface; + bbl_session_s *session; + bbl_l2tp_session_t *l2tp_session; + + uint8_t *buf; + uint16_t tx_len; + uint16_t rx_len; + uint64_t rx_first_seq; + uint64_t rx_last_seq; + + uint64_t tx_interval; /* TX interval in nsec */ + uint64_t send_window_packets; + struct timespec send_window_start; + + uint8_t rx_priority; /* IPv4 TOS or IPv6 TC */ + uint8_t rx_outer_vlan_pbit; + uint8_t rx_inner_vlan_pbit; + + uint64_t packets_tx; + uint64_t packets_rx; + uint64_t loss; + + uint64_t min_delay_ns; + uint64_t max_delay_ns; + + bbl_rate_s rate_packets_tx; + bbl_rate_s rate_packets_rx; + + void *next; /* next stream of same session */ +} bbl_stream; + +bool +bbl_stream_add(bbl_ctx_s *ctx, bbl_access_config_s *access_config, bbl_session_s *session); + +void +bbl_stream_tx_job (timer_s *timer); + +#endif \ No newline at end of file diff --git a/src/bbl_timer.c b/src/bbl_timer.c index a5338206..80f1e421 100644 --- a/src/bbl_timer.c +++ b/src/bbl_timer.c @@ -50,13 +50,13 @@ timespec_sub (struct timespec *result, struct timespec *x, struct timespec *y) /* * Avoid overflow of result->tv_nsec */ - if (x->tv_nsec < y->tv_nsec) { - result->tv_nsec = x->tv_nsec + 1e9 - y->tv_nsec; - result->tv_sec = x->tv_sec - y->tv_sec - 1; - } else { - result->tv_sec = x->tv_sec - y->tv_sec; - result->tv_nsec = x->tv_nsec - y->tv_nsec; - } + if (x->tv_nsec < y->tv_nsec) { + result->tv_nsec = x->tv_nsec + 1e9 - y->tv_nsec; + result->tv_sec = x->tv_sec - y->tv_sec - 1; + } else { + result->tv_sec = x->tv_sec - y->tv_sec; + result->tv_nsec = x->tv_nsec - y->tv_nsec; + } } /* diff --git a/src/bbl_utils.h b/src/bbl_utils.h index 1177a775..45df99aa 100644 --- a/src/bbl_utils.h +++ b/src/bbl_utils.h @@ -23,5 +23,4 @@ char *format_ipv6_prefix(ipv6_prefix *addr6); char *replace_substring (const char* s, const char* old, const char* new); const char *val2key (struct keyval_ *keyval, uint val); - #endif