mirror/rtbrick-bngblaster
1
0
Fork 0
mirror of https://github.com/rtbrick/bngblaster.git synced 2024-05-06 15:54:57 +00:00
Code Issues Projects Releases Wiki Activity

Traffic Streams (WIP)

Browse Source
This commit is contained in:
Christian Giese
2021-03-30 13:23:36 +02:00
parent e62164f987
commit 67f8182ef9
20 changed files with 1278 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bbl.c
+6 -3
View File
@@ -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");
src/bbl.h
+17 -3
View File
@@ -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;
src/bbl_config.c
+112 -8
View File
@@ -9,6 +9,7 @@
#include "bbl.h"
#include "bbl_config.h"
#include "bbl_stream.h"
#include <jansson.h>
#include <sys/stat.h>
@@ -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;
src/bbl_ctrl.c
+67
View File
@@ -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},
};
src/bbl_ctx.c
+22 -21
View File
@@ -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;
}
src/bbl_io_netmap.c
+22 -4
View File
@@ -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, &eth);
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;
}
src/bbl_io_netmap.h
+3
View File
@@ -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);
src/bbl_io_packet_mmap.c
+61 -8
View File
@@ -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;
}
src/bbl_io_packet_mmap.h
+3
View File
@@ -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);
src/bbl_io_raw.c
+27 -10
View File
@@ -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, &eth);
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;
}
src/bbl_io_raw.h
+3
View File
@@ -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);
src/bbl_protocols.c
+10 -4
View File
@@ -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));
src/bbl_protocols.h
+3 -3
View File
@@ -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_ {
src/bbl_rx.c
+177 -128
View File
@@ -9,6 +9,7 @@
#include "bbl.h"
#include "bbl_session.h"
#include "bbl_stream.h"
#include <openssl/md5.h>
#include <openssl/rand.h>
@@ -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, &eth->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;
src/bbl_session.c
+7
View File
@@ -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++;
src/bbl_stats.h
+1
View File
@@ -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);
src/bbl_stream.c
+638
View File
@@ -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, &eth) != 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, &eth) != 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, &eth) != 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, &eth) != 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;
}
src/bbl_stream.h
+92
View File
@@ -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
src/bbl_timer.c
+7 -7
View File
@@ -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;
}
}
/*
src/bbl_utils.h
-1
View File
@@ -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