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

Add LACP (initial working draft)

Browse Source
This commit is contained in:
Christian Giese
2022-11-02 17:52:51 +01:00
committed by Christian Giese
parent 1271bf86ac
commit 71f87fbeeb
29 changed files with 697 additions and 523 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/bngblaster/src/bbl.c
+9 -21
View File
@@ -122,6 +122,7 @@ struct keyval_ log_names[] = {
{ ISIS, "isis" },
{ BGP, "bgp" },
{ TCP, "tcp" },
{ LAG, "lag" },
{ 0, NULL}
};
@@ -462,10 +463,8 @@ main(int argc, char *argv[])
goto CLEANUP;
}
#if 0
timer_test(ctx);
exit(0);
#endif
/* Open logfile. */
log_open();
/* Init config. */
bbl_config_init_defaults();
@@ -503,21 +502,14 @@ main(int argc, char *argv[])
}
#endif
/* Init curses. */
if (interactive) {
bbl_interactive_init();
}
/* Init IS-IS instances. */
if(!isis_init()) {
if(interactive) endwin();
fprintf(stderr, "Error: Failed to init IS-IS\n");
goto CLEANUP;
}
/* Init interfaces. */
if(!bbl_interface_init()) {
if(interactive) endwin();
fprintf(stderr, "Error: Failed to init interfaces\n");
goto CLEANUP;
}
@@ -527,14 +519,12 @@ main(int argc, char *argv[])
/* Init BGP sessions. */
if(!bgp_init()) {
if(interactive) endwin();
fprintf(stderr, "Error: Failed to init BGP\n");
goto CLEANUP;
}
/* Init streams. */
if(!bbl_stream_init()) {
if(interactive) endwin();
fprintf(stderr, "Error: Failed to add RAW stream traffic\n");
goto CLEANUP;
}
@@ -545,17 +535,11 @@ main(int argc, char *argv[])
/* Setup test. */
if(bbl_access_interface_get(NULL)) {
if(!bbl_sessions_init()) {
if(interactive) endwin();
fprintf(stderr, "Error: Failed to init sessions\n");
goto CLEANUP;
}
}
/* Start curses. */
if(interactive) {
bbl_interactive_start();
}
/* Setup control job. */
timer_add_periodic(&g_ctx->timer_root, &g_ctx->control_timer, "Control Timer",
1, 0, g_ctx, &bbl_ctrl_job);
@@ -563,7 +547,6 @@ main(int argc, char *argv[])
/* Setup control socket and job */
if(g_ctx->ctrl_socket_path) {
if(!bbl_ctrl_socket_init()) {
if(interactive) endwin();
goto CLEANUP;
}
}
@@ -584,8 +567,13 @@ main(int argc, char *argv[])
/* Start threads. */
io_thread_start_all(g_ctx);
/* Start curses. */
if(interactive) {
bbl_interactive_init();
bbl_interactive_start();
}
/* Start event loop. */
log_open();
clock_gettime(CLOCK_MONOTONIC, &g_ctx->timestamp_start);
signal(SIGINT, teardown_handler);
while(g_teardown_request_count < 10) {
code/bngblaster/src/bbl_a10nsp.c
+5 -2
View File
@@ -54,11 +54,14 @@ bbl_a10nsp_interfaces_add()
a10nsp_interface = calloc(1, sizeof(bbl_a10nsp_interface_s));
interface->a10nsp = a10nsp_interface;
a10nsp_interface->interface = interface;
a10nsp_config->a10nsp_interface = a10nsp_interface;
CIRCLEQ_INSERT_TAIL(&g_ctx->a10nsp_interface_qhead, a10nsp_interface, a10nsp_interface_qnode);
/* Init interface */
a10nsp_interface->name = a10nsp_config->interface;
a10nsp_interface->interface = interface;
/* Init TXQ */
a10nsp_interface->txq = calloc(1, sizeof(bbl_txq_s));
bbl_txq_init(a10nsp_interface->txq, BBL_TXQ_DEFAULT_SIZE);
@@ -481,7 +484,7 @@ bbl_a10nsp_rx(bbl_a10nsp_interface_s *interface,
/* Create A10NSP session if not already present */
if(!session->a10nsp_session) {
LOG(DEBUG, "A10NSP (ID: %u) Session created on interface %s with S-VLAN %d\n",
session->session_id, interface->interface->name, eth->vlan_outer);
session->session_id, interface->name, eth->vlan_outer);
session->a10nsp_session = calloc(1, sizeof(bbl_a10nsp_session_s));
session->a10nsp_session->session = session;
session->a10nsp_session->a10nsp_interface = interface;
code/bngblaster/src/bbl_a10nsp.h
+7 -2
View File
@@ -21,8 +21,14 @@
typedef struct bbl_a10nsp_interface_
{
bbl_interface_s *interface;
char *name; /* interface name */
uint32_t ifindex; /* interface index */
/* parent */
bbl_interface_s *interface;
bbl_txq_s *txq;
uint8_t mac[ETH_ADDR_LEN];
bool qinq;
@@ -68,7 +74,6 @@ typedef struct bbl_a10nsp_interface_
CIRCLEQ_ENTRY(bbl_a10nsp_interface_) a10nsp_interface_qnode;
CIRCLEQ_HEAD(session_tx_a10nsp_, bbl_session_ ) session_tx_qhead; /* list of sessions that want to transmit */
} bbl_a10nsp_interface_s;
typedef struct bbl_a10nsp_session_
code/bngblaster/src/bbl_access.c
+15 -5
View File
@@ -53,6 +53,8 @@ bbl_access_interfaces_add()
bbl_access_interface_s *access_interface;
bbl_interface_s *interface;
char ifname[SUB_STR_LEN];
while(access_config) {
interface = bbl_interface_get(access_config->interface);
if(!interface) {
@@ -60,12 +62,17 @@ bbl_access_interfaces_add()
return false;
}
if(!interface->access) {
snprintf(ifname, sizeof(ifname), "%s", access_config->interface);
access_interface = calloc(1, sizeof(bbl_access_interface_s));
interface->access = access_interface;
access_interface->interface = interface;
CIRCLEQ_INSERT_TAIL(&g_ctx->access_interface_qhead, access_interface, access_interface_qnode);
/* Init interface */
access_interface->name = access_config->interface;
access_interface->interface = interface;
/* Init TXQ */
access_interface->txq = calloc(1, sizeof(bbl_txq_s));
bbl_txq_init(access_interface->txq, BBL_TXQ_DEFAULT_SIZE);
@@ -114,7 +121,8 @@ bbl_access_interface_get(char *interface_name)
static void
bbl_access_update_eth(bbl_session_s *session,
bbl_ethernet_header_s *eth) {
bbl_ethernet_header_s *eth)
{
uint8_t *dst = eth->dst;
eth->dst = eth->src;
eth->src = dst;
@@ -130,7 +138,8 @@ static bbl_txq_result_t
bbl_access_icmp_reply(bbl_session_s *session,
bbl_ethernet_header_s *eth,
bbl_ipv4_s *ipv4,
bbl_icmp_s *icmp) {
bbl_icmp_s *icmp)
{
uint32_t dst = ipv4->dst;
bbl_access_update_eth(session, eth);
ipv4->dst = ipv4->src;
@@ -144,7 +153,8 @@ static bbl_txq_result_t
bbl_access_icmpv6_na(bbl_session_s *session,
bbl_ethernet_header_s *eth,
bbl_ipv6_s *ipv6,
bbl_icmpv6_s *icmpv6) {
bbl_icmpv6_s *icmpv6)
{
bbl_access_update_eth(session, eth);
ipv6->dst = ipv6->src;
ipv6->src = icmpv6->prefix.address;
@@ -1648,7 +1658,7 @@ bbl_access_session_id_from_vlan(bbl_access_interface_s *interface,
bbl_session_s *session;
void **search;
key.ifindex = interface->interface->ifindex;
key.ifindex = interface->ifindex;
key.outer_vlan_id = eth->vlan_outer;
key.inner_vlan_id = eth->vlan_inner;
code/bngblaster/src/bbl_access.h
+6 -1
View File
@@ -12,7 +12,12 @@
typedef struct bbl_access_interface_
{
bbl_interface_s *interface;
char *name; /* interface name */
uint32_t ifindex; /* interface index */
/* parent */
bbl_interface_s *interface;
bbl_txq_s *txq;
uint8_t mac[ETH_ADDR_LEN];
code/bngblaster/src/bbl_config.c
+89 -53
View File
@@ -242,12 +242,15 @@ json_parse_lag(json_t *lag, bbl_lag_config_s *lag_config)
{
json_t *value = NULL;
const char *s = NULL;
static uint8_t lag_id = 0;
value = json_object_get(lag, "lag-id");
if(value) {
lag_config->id = json_number_value(value);
lag_config->id = ++lag_id;
if(json_unpack(lag, "{s:s}", "interface", &s) == 0) {
lag_config->interface = strdup(s);
} else {
fprintf(stderr, "JSON config error: Missing value for lag->id\n");
fprintf(stderr, "JSON config error: Missing value for lag->interface\n");
return false;
}
value = json_object_get(lag, "lacp");
if(json_is_boolean(value)) {
@@ -277,20 +280,55 @@ json_parse_lag(json_t *lag, bbl_lag_config_s *lag_config)
} else {
lag_config->lacp_system_id[0] = 0x02;
lag_config->lacp_system_id[1] = 0xff;
lag_config->lacp_system_id[5] = lag_config->id;
lag_config->lacp_system_id[2] = 0xff;
lag_config->lacp_system_id[3] = 0xff;
lag_config->lacp_system_id[4] = 0xff;
}
if(json_unpack(lag, "{s:s}", "mac", &s) == 0) {
if(sscanf(s, "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx",
&lag_config->mac[0],
&lag_config->mac[1],
&lag_config->mac[2],
&lag_config->mac[3],
&lag_config->mac[4],
&lag_config->mac[5]) < 6) {
fprintf(stderr, "JSON config error: Invalid value for lag->mac\n");
return false;
}
} else {
lag_config->mac[0] = 0x02;
lag_config->mac[1] = 0xff;
lag_config->mac[2] = 0xff;
lag_config->mac[3] = 0xff;
lag_config->mac[4] = 0xff;
lag_config->mac[5] = lag_config->id;
}
return true;
}
static bool
link_present(char *interface_name)
lag_present(char *interface)
{
bbl_link_config_s *link_config = g_ctx->config.link_config;
while(link_config) {
if(strcmp(link_config->interface, interface_name) == 0) {
bbl_lag_config_s *config = g_ctx->config.lag_config;
while(config) {
if(strcmp(config->interface, interface) == 0) {
return true;
}
link_config = link_config->next;
config = config->next;
}
return false;
}
static bool
link_present(char *interface)
{
bbl_link_config_s *config = g_ctx->config.link_config;
while(config) {
if(config->interface && strcmp(config->interface, interface) == 0) {
return true;
}
config = config->next;
}
return false;
}
@@ -299,8 +337,7 @@ static void
link_add(char *interface_name)
{
bbl_link_config_s *link_config;
if(link_present(interface_name)) {
if(link_present(interface_name)|| lag_present(interface_name)) {
return;
}
@@ -322,14 +359,19 @@ static bool
json_parse_link(json_t *link, bbl_link_config_s *link_config)
{
json_t *value = NULL;
const char *s = NULL;
char *s = NULL;
if(json_unpack(link, "{s:s}", "interface", &s) == 0) {
if(link_present(s) || lag_present(s)) {
fprintf(stderr, "JSON config error: Duplicate link configuration for %s\n", s);
return false;
}
link_config->interface = strdup(s);
} else {
fprintf(stderr, "JSON config error: Missing value for links->interface\n");
return false;
}
if(json_unpack(link, "{s:s}", "description", &s) == 0) {
link_config->description = strdup(s);
}
@@ -340,8 +382,7 @@ json_parse_link(json_t *link, bbl_link_config_s *link_config)
&link_config->mac[2],
&link_config->mac[3],
&link_config->mac[4],
&link_config->mac[5]) < 6)
{
&link_config->mac[5]) < 6) {
fprintf(stderr, "JSON config error: Invalid value for links->mac\n");
return false;
}
@@ -410,15 +451,20 @@ json_parse_link(json_t *link, bbl_link_config_s *link_config)
} else {
link_config->rx_threads = g_ctx->config.rx_threads;
}
value = json_object_get(link, "lag-id");
if(value) {
link_config->lag_id = json_number_value(value);
}
value = json_object_get(link, "lacp-priority");
if(value) {
link_config->lacp_priority = json_number_value(value);
} else {
link_config->lacp_priority = 32768;
/* Link Aggregation Group (LAG) Configuration */
if(json_unpack(link, "{s:s}", "lag-interface", &s) == 0) {
if(!lag_present(s)) {
fprintf(stderr, "JSON config error: Missing configuration for lag-interface %s\n", s);
return false;
}
link_config->lag_interface = strdup(s);
value = json_object_get(link, "lacp-priority");
if(value) {
link_config->lacp_priority = json_number_value(value);
} else {
link_config->lacp_priority = 32768;
}
}
return true;
}
@@ -542,6 +588,24 @@ json_parse_access_interface(json_t *access_interface, bbl_access_config_s *acces
uint32_t ipv4;
double number;
if(json_unpack(access_interface, "{s:s}", "interface", &s) == 0) {
access_config->interface = strdup(s);
link_add(access_config->interface);
} else {
fprintf(stderr, "JSON config error: Missing value for interface->interface\n");
return false;
}
if(json_unpack(access_interface, "{s:s}", "network-interface", &s) == 0) {
access_config->network_interface = strdup(s);
}
if(json_unpack(access_interface, "{s:s}", "a10nsp-interface", &s) == 0) {
if(access_config->network_interface) {
fprintf(stderr, "JSON config error: You can't define access->network-interface and access->a10nsp-interface\n");
return false;
}
access_config->a10nsp_interface = strdup(s);
}
value = json_object_get(access_interface, "i1-start");
if(value) {
access_config->i1 = json_number_value(value);
@@ -587,24 +651,6 @@ json_parse_access_interface(json_t *access_interface, bbl_access_config_s *acces
return false;
}
}
if(json_unpack(access_interface, "{s:s}", "interface", &s) == 0) {
access_config->interface = strdup(s);
link_add(access_config->interface);
} else {
fprintf(stderr, "JSON config error: Missing value for access->interface\n");
return false;
}
if(json_unpack(access_interface, "{s:s}", "network-interface", &s) == 0) {
access_config->network_interface = strdup(s);
}
if(json_unpack(access_interface, "{s:s}", "a10nsp-interface", &s) == 0) {
if(access_config->network_interface) {
fprintf(stderr, "JSON config error: You can't define access->network-interface and access->a10nsp-interface\n");
return false;
}
access_config->a10nsp_interface = strdup(s);
}
value = json_object_get(access_interface, "monkey");
if(json_is_boolean(value)) {
@@ -651,7 +697,7 @@ json_parse_access_interface(json_t *access_interface, bbl_access_config_s *acces
}
if(access_config->access_outer_vlan_min > access_config->access_outer_vlan_max ||
access_config->access_inner_vlan_min > access_config->access_inner_vlan_max) {
fprintf(stderr, "JSON config error: Invalid VLAN range (min > max)\n");
fprintf(stderr, "JSON config error: Invalid value for access VLAN range (min > max)\n");
return false;
}
value = json_object_get(access_interface, "third-vlan");
@@ -1989,15 +2035,6 @@ json_parse_config(json_t *root)
}
g_ctx->config.stream_max_per_group = number;
}
value = json_object_get(section, "stream-precision");
if(json_is_number(value)) {
number = json_number_value(value);
if(number < 1 || number > UINT8_MAX) {
fprintf(stderr, "JSON config error: Invalid value for traffic->stream-precision\n");
return false;
}
g_ctx->config.stream_tx_precision = number;
}
value = json_object_get(section, "stream-rate-calculation");
if(json_is_boolean(value)) {
g_ctx->config.stream_rate_calc = json_boolean_value(value);
@@ -2605,6 +2642,5 @@ bbl_config_init_defaults()
g_ctx->config.stream_rate_calc = true;
g_ctx->config.stream_max_ppi = 16;
g_ctx->config.stream_max_per_group = 64;
g_ctx->config.stream_tx_precision = 2;
g_ctx->config.session_traffic_autostart = true;
}
code/bngblaster/src/bbl_config.h
+8 -3
View File
@@ -14,6 +14,7 @@ typedef struct bbl_access_config_
bool exhausted;
uint32_t sessions; /* per access config session counter */
uint8_t lag_id;
char *interface;
char *network_interface;
char *a10nsp_interface;
@@ -86,7 +87,7 @@ typedef struct bbl_access_config_
typedef struct bbl_network_config_
{
char *interface;
uint8_t mac[ETH_ADDR_LEN];
uint8_t gateway_mac[ETH_ADDR_LEN];
bool gateway_resolve_wait;
@@ -112,6 +113,7 @@ typedef struct bbl_network_config_
typedef struct bbl_a10nsp_config_
{
char *interface;
uint8_t mac[ETH_ADDR_LEN];
bool qinq;
@@ -137,8 +139,8 @@ typedef struct bbl_link_config_
uint8_t tx_threads;
uint8_t rx_threads;
uint8_t lag_id;
char *lag_interface;
uint32_t lacp_priority;
void *next; /* pointer to next link config element */
@@ -148,10 +150,13 @@ typedef struct bbl_link_config_
typedef struct bbl_lag_config_
{
uint8_t id;
char *interface;
bool lacp_enable;
bool lacp_timeout_short;
uint32_t lacp_system_priority;
uint16_t lacp_system_priority;
uint8_t lacp_system_id[ETH_ADDR_LEN];
uint8_t mac[ETH_ADDR_LEN];
void *next; /* pointer to next lag config element */
} bbl_lag_config_s;
code/bngblaster/src/bbl_ctrl.c
+6 -6
View File
@@ -663,7 +663,7 @@ bbl_ctrl_network_interface_json(bbl_network_interface_s *interface)
{
return json_pack("{ss si ss si si si si si si si si si si si si si si si si si si si si si si si si si si si si si si}",
"name", interface->name,
"ifindex", interface->interface->ifindex,
"ifindex", interface->ifindex,
"type", "network",
"tx-packets", interface->stats.packets_tx,
"tx-bytes", interface->stats.bytes_tx,
@@ -702,8 +702,8 @@ static json_t *
bbl_ctrl_access_interface_json(bbl_access_interface_s *interface)
{
return json_pack("{ss si ss si si si si si si si si si si si si si si si si si si si si si si si si si si si si si si si}",
"name", interface->interface->name,
"ifindex", interface->interface->ifindex,
"name", interface->name,
"ifindex", interface->ifindex,
"type", "access",
"tx-packets", interface->stats.packets_tx,
"tx-bytes", interface->stats.bytes_tx,
@@ -743,8 +743,8 @@ static json_t *
bbl_ctrl_a10nsp_interface_json(bbl_a10nsp_interface_s *interface)
{
return json_pack("{ss si ss si si si si si si si si si si si si si si si si si si si si si si si si si si si si}",
"name", interface->interface->name,
"ifindex", interface->interface->ifindex,
"name", interface->name,
"ifindex", interface->ifindex,
"type", "a10nsp",
"tx-packets", interface->stats.packets_tx,
"tx-bytes", interface->stats.bytes_tx,
@@ -1785,7 +1785,7 @@ bbl_ctrl_socket_thread(void *thread_data)
/* Use first interface as default. */
access_interface = bbl_access_interface_get(NULL);
if(access_interface) {
key.ifindex = access_interface->interface->ifindex;
key.ifindex = access_interface->ifindex;
}
}
value = json_object_get(arguments, "outer-vlan");
code/bngblaster/src/bbl_ctx.h
+2 -3
View File
@@ -78,14 +78,14 @@ typedef struct bbl_ctx_
dict *li_flow_dict; /* hashtable for LI flows */
dict *stream_flow_dict; /* hashtable for traffic stream flows */
bbl_stream_group_s *stream_groups;
uint16_t next_tunnel_id;
uint64_t flow_id;
char *ctrl_socket_path;
bbl_ctrl_thread_s *ctrl_thread;
void *stream_thread; /* single linked list of threads */
io_thread_s *io_threads; /* single linked list of threads */
bool tcp;
@@ -313,7 +313,6 @@ typedef struct bbl_ctx_
bool stream_rate_calc; /* Enable/disable stream rate calculation */
uint8_t stream_max_ppi; /* Limit max packets per interval */
uint8_t stream_max_per_group; /* Limit max streams per group */
uint8_t stream_tx_precision; /* Set stream precision */
/* Session Traffic */
bool session_traffic_autostart;
code/bngblaster/src/bbl_def.h
+16 -1
View File
@@ -114,9 +114,17 @@ typedef enum {
typedef enum {
INTERFACE_DISABLED = 0,
INTERFACE_UP,
INTERFACE_DOWN
INTERFACE_DOWN,
INTERFACE_STANDBY,
} __attribute__ ((__packed__)) interface_state_t;
typedef enum {
DEFAULT_INTERFACE = 0,
LAG_INTERFACE,
LAG_MEMBER_INTERFACE,
} __attribute__ ((__packed__)) interface_type_t;
typedef enum {
IGMP_GROUP_IDLE = 0,
IGMP_GROUP_LEAVING,
@@ -131,6 +139,13 @@ typedef enum {
ENDPOINT_ACTIVE,
} __attribute__ ((__packed__)) endpoint_state_t;
typedef enum {
LACP_DISABLED = 0,
LACP_EXPIRED,
LACP_DEFAULTED,
LACP_CURRENT
} __attribute__ ((__packed__)) lacp_state_t;
/*
* Session state
*/
code/bngblaster/src/bbl_interactive.c
+6 -6
View File
@@ -423,10 +423,10 @@ bbl_interactive_window_job(timer_s *timer)
CIRCLEQ_FOREACH(a10nsp_if, &g_ctx->a10nsp_interface_qhead, a10nsp_interface_qnode) {
if(a10nsp_if == g_a10nsp_if) {
wattron(stats_win, COLOR_PAIR(COLOR_GREEN));
wprintw(stats_win, " %s", a10nsp_if->interface->name);
wprintw(stats_win, " %s", a10nsp_if->name);
wattroff(stats_win, COLOR_PAIR(COLOR_GREEN));
} else {
wprintw(stats_win, " %s", a10nsp_if->interface->name);
wprintw(stats_win, " %s", a10nsp_if->name);
}
}
wprintw(stats_win, " )\n TX Packets %10lu |%7lu PPS %10lu Kbps\n",
@@ -465,10 +465,10 @@ bbl_interactive_window_job(timer_s *timer)
CIRCLEQ_FOREACH(access_if, &g_ctx->access_interface_qhead, access_interface_qnode) {
if(access_if == g_access_if) {
wattron(stats_win, COLOR_PAIR(COLOR_GREEN));
wprintw(stats_win, " %s", access_if->interface->name);
wprintw(stats_win, " %s", access_if->name);
wattroff(stats_win, COLOR_PAIR(COLOR_GREEN));
} else {
wprintw(stats_win, " %s", access_if->interface->name);
wprintw(stats_win, " %s", access_if->name);
}
}
wprintw(stats_win, " )\n TX Packets %10lu |%7lu PPS %10lu Kbps\n",
@@ -511,10 +511,10 @@ bbl_interactive_window_job(timer_s *timer)
CIRCLEQ_FOREACH(access_if, &g_ctx->access_interface_qhead, access_interface_qnode) {
if(access_if == g_access_if) {
wattron(stats_win, COLOR_PAIR(COLOR_GREEN));
wprintw(stats_win, " %s", access_if->interface->name);
wprintw(stats_win, " %s", access_if->name);
wattroff(stats_win, COLOR_PAIR(COLOR_GREEN));
} else {
wprintw(stats_win, " %s", access_if->interface->name);
wprintw(stats_win, " %s", access_if->name);
}
}
wprintw(stats_win, " )\n");
code/bngblaster/src/bbl_interface.c
+30 -2
View File
@@ -9,6 +9,29 @@
#include "bbl.h"
#include <sys/stat.h>
const char *
interface_type_string(interface_type_t type)
{
switch(type) {
case DEFAULT_INTERFACE: return "Interface";
case LAG_INTERFACE: return "LAG-Interface";
case LAG_MEMBER_INTERFACE: return "LAG-Member-Interface";
default: return "N/A";
}
}
const char *
interface_state_string(interface_state_t state)
{
switch(state) {
case INTERFACE_DISABLED: return "Disabled";
case INTERFACE_UP: return "Up";
case INTERFACE_DOWN: return "Down";
case INTERFACE_STANDBY: return "Standby";
default: return "N/A";
}
}
/**
* bbl_interface_lock
*
@@ -101,10 +124,10 @@ bbl_interface_link_add(char *interface_name, bbl_link_config_s *link_config)
CIRCLEQ_INSERT_TAIL(&g_ctx->interface_qhead, interface, interface_qnode);
interface->config = link_config;
if(!bbl_lag_interface_add(interface, link_config)) {
if(!io_interface_init(interface)) {
return NULL;
}
if(!io_interface_init(interface)) {
if(!bbl_lag_interface_add(interface, link_config)) {
return NULL;
}
return interface;
@@ -179,6 +202,11 @@ bbl_interface_s *
bbl_interface_get(char *interface_name)
{
bbl_interface_s *interface;
if(!interface_name) {
return NULL;
}
CIRCLEQ_FOREACH(interface, &g_ctx->interface_qhead, interface_qnode) {
if (strcmp(interface->name, interface_name) == 0) {
return interface;
code/bngblaster/src/bbl_interface.h
+16 -11
View File
@@ -13,33 +13,38 @@
typedef struct bbl_interface_
{
char *name; /* interface name */
interface_type_t type; /* interface type */
interface_state_t state; /* interface state */
uint32_t ifindex; /* interface index */
uint32_t pcap_index; /* interface index for packet captures */
bbl_link_config_s *config;
bbl_lag_member_s *lag;
bbl_a10nsp_interface_s *a10nsp;
bbl_lag_s *lag;
bbl_lag_member_s *lag_member;
bbl_access_interface_s *access;
bbl_network_interface_s *network;
bbl_a10nsp_interface_s *a10nsp;
interface_state_t state;
uint8_t mac[ETH_ADDR_LEN];
uint32_t send_requests;
CIRCLEQ_ENTRY(bbl_interface_) interface_qnode;
CIRCLEQ_ENTRY(bbl_interface_) interface_lag_qnode;
struct {
struct timer_ *rx_job;
struct timer_ *tx_job;
io_handle_s *rx;
io_handle_s *tx;
} io;
uint32_t ifindex; /* interface index */
uint32_t pcap_index; /* interface index for packet captures */
struct timer_ *tx_job;
struct timer_ *rx_job;
} bbl_interface_s;
const char *
interface_type_string(interface_type_t type);
const char *
interface_state_string(interface_state_t state);
void
bbl_interface_unlock_all();
code/bngblaster/src/bbl_lag.c
+102 -51
View File
@@ -10,20 +10,12 @@
uint16_t g_lag_port_id = 1;
/**
* bbl_lag_get
*
* Get interface by name.
*
* @param id LAG identifier
* @return the LAG group or NULL
*/
bbl_lag_s *
bbl_lag_get(uint8_t id)
bbl_lag_get_by_name(char *interface)
{
bbl_lag_s *lag;
CIRCLEQ_FOREACH(lag, &g_ctx->lag_qhead, lag_qnode) {
if(lag->id == id) {
if(strcmp(lag->interface->name, interface) == 0) {
return lag;
}
}
@@ -42,44 +34,57 @@ bbl_lag_get(uint8_t id)
bool
bbl_lag_add()
{
bbl_lag_config_s *lag_config = g_ctx->config.lag_config;
bbl_lag_config_s *config = g_ctx->config.lag_config;
bbl_interface_s *interface;
bbl_lag_s *lag;
char name[sizeof("lag255")];
while(lag_config) {
snprintf(name, sizeof(name), "lag%u", lag->id);
CIRCLEQ_FOREACH(lag, &g_ctx->lag_qhead, lag_qnode) {
if(lag->id == lag_config->id) {
LOG(ERROR, "Failed to add %s (duplicate)\n", name);
return false;
}
}
while(config) {
lag = calloc(1, sizeof(bbl_lag_s));
lag->id = lag_config->id;
lag->interface = strdup(name);
lag->config = lag_config;
CIRCLEQ_INIT(&lag->lag_interface_qhead);
CIRCLEQ_INSERT_TAIL(&g_ctx->lag_qhead, lag, lag_qnode);
if(!lag) return false;
lag->id = config->id;
lag->config = config;
lag_config = lag_config->next;
interface = calloc(1, sizeof(bbl_interface_s));
if(!interface) return false;
lag->interface = interface;
interface->name = config->interface;
interface->type = LAG_INTERFACE;
interface->state = INTERFACE_DOWN;
interface->ifindex = lag->id;
interface->pcap_index = g_ctx->pcap.index++;
interface->lag = lag;
memcpy(interface->mac, config->mac, ETH_ADDR_LEN);
CIRCLEQ_INIT(&lag->lag_member_qhead);
CIRCLEQ_INSERT_TAIL(&g_ctx->lag_qhead, lag, lag_qnode);
CIRCLEQ_INSERT_TAIL(&g_ctx->interface_qhead, interface, interface_qnode);
LOG(LAG, "LAG (%s) New lag-interface created\n", interface->name);
config = config->next;
}
return true;
}
void
bbl_lag_lacp_tx_job(timer_s *timer)
bbl_lag_lacp_job(timer_s *timer)
{
bbl_interface_s *interface = timer->data;
bbl_lag_member_s *member = interface->lag;
bbl_lag_member_s *member = interface->lag_member;
interface->send_requests |= BBL_SEND_LACP;
member->timeout++;
if(member->timeout > 3) {
member->state = INTERFACE_DOWN;
interface->state = INTERFACE_DOWN;
if(!(member->actor_state & LACP_STATE_FLAG_EXPIRED)) {
member->actor_state |= LACP_STATE_FLAG_EXPIRED;
LOG(LAG, "LAG (%s) LACP expired on interface %s\n",
member->lag->interface->name, interface->name);
}
if(member->timeout > 6) {
if(!(member->actor_state & LACP_STATE_FLAG_DEFAULTED)) {
member->actor_state |= LACP_STATE_FLAG_DEFAULTED;
}
}
}
}
@@ -90,24 +95,35 @@ bbl_lag_interface_add(bbl_interface_s *interface, bbl_link_config_s *link_config
bbl_lag_member_s *member;
time_t timer_sec;
if(link_config->lag_id) {
lag = bbl_lag_get(link_config->lag_id);
if(link_config->lag_interface) {
lag = bbl_lag_get_by_name(link_config->lag_interface);
if(!lag) {
LOG(ERROR, "Failed to add link %s (LAG %u not defined)\n",
link_config->interface, link_config->lag_id);
LOG(ERROR, "Failed to add link %s to lag-interface %s (not found)\n",
link_config->interface, link_config->lag_interface);
return false;
}
if(link_config->tx_threads) {
LOG(ERROR, "Failed to add link %s to lag-interface %s (TX threads not allowed for LAG interfaces)\n",
link_config->interface, link_config->lag_interface);
return false;
}
member = calloc(1, sizeof(bbl_lag_member_s));
member->lag = lag;
member->interface = interface;
interface->type = LAG_MEMBER_INTERFACE;
interface->lag = lag;
interface->lag_member = member;
memcpy(interface->mac, lag->interface->mac, ETH_ADDR_LEN);
if(lag->config->lacp_enable) {
member->state = INTERFACE_DOWN;
member->lacp_state = LACP_DEFAULTED;
interface->state = INTERFACE_DOWN;
memcpy(member->actor_system_id, lag->config->lacp_system_id, ETH_ADDR_LEN);
member->actor_system_priority = lag->config->lacp_system_priority;
member->actor_key = lag->id;
member->actor_port_priority = interface->config->lacp_priority;
member->actor_port_id = g_lag_port_id++;
member->actor_state = LACP_STATE_FLAG_ACTIVE|LACP_STATE_FLAG_IN_SYNC|LACP_STATE_FLAG_AGGREGATION|LACP_STATE_FLAG_DEFAULTED;
member->actor_state = LACP_STATE_FLAG_ACTIVE|LACP_STATE_FLAG_IN_SYNC|LACP_STATE_FLAG_COLLECTING|LACP_STATE_FLAG_AGGREGATION|LACP_STATE_FLAG_DEFAULTED;
if(lag->config->lacp_timeout_short) {
timer_sec = 1;
member->actor_state |= LACP_STATE_FLAG_SHORT_TIMEOUT;
@@ -115,32 +131,67 @@ bbl_lag_interface_add(bbl_interface_s *interface, bbl_link_config_s *link_config
timer_sec = 30;
}
timer_add_periodic(&g_ctx->timer_root, &member->lacp_timer, "LACP",
timer_sec, 0, interface, &bbl_lag_lacp_tx_job);
timer_sec, 0, interface, &bbl_lag_lacp_job);
} else {
member->state = INTERFACE_UP;
member->lacp_state = LACP_DISABLED;
}
interface->lag = member;
CIRCLEQ_INSERT_TAIL(&lag->lag_interface_qhead, interface, interface_lag_qnode);
CIRCLEQ_INSERT_TAIL(&lag->lag_member_qhead, member, lag_member_qnode);
LOG(LAG, "LAG (%s) Interface %s added\n", lag->interface->name, interface->name);
}
return true;
}
static void
bbl_lag_select(bbl_lag_s *lag)
{
bbl_lag_member_s *member;
lag->active_count = 0;
CIRCLEQ_FOREACH(member, &lag->lag_member_qhead, lag_member_qnode) {
member->primary = false;
if(member->partner_state & (LACP_STATE_FLAG_COLLECTING|LACP_STATE_FLAG_DISTRIBUTING)) {
member->actor_state |= (LACP_STATE_FLAG_COLLECTING|LACP_STATE_FLAG_DISTRIBUTING);
member->interface->state = INTERFACE_UP;
lag->active_list[lag->active_count++] = member;
if(lag->active_count == 1) {
member->primary = true;
LOG(LAG, "LAG (%s) Interface %s set to UP (primary)\n", member->lag->interface->name, member->interface->name);
} else {
LOG(LAG, "LAG (%s) Interface %s set to UP\n", member->lag->interface->name, member->interface->name);
}
} else {
member->interface->state = INTERFACE_DOWN;
}
}
if(lag->active_count) {
lag->interface->state = INTERFACE_UP;
} else {
lag->interface->state = INTERFACE_DOWN;
}
}
void
bbl_lag_rx_lacp(bbl_interface_s *interface,
bbl_ethernet_header_s *eth)
{
bbl_lag_member_s *member = interface->lag;
bbl_lag_member_s *member = interface->lag_member;
bbl_lacp_s *lacp = (bbl_lacp_s*)eth->next;
if(member) {
memcpy(member->partner_system_id, lacp->actor_system_id, ETH_ADDR_LEN);
member->partner_system_priority = lacp->actor_system_priority;
member->partner_key = lacp->actor_key;
member->partner_port_priority = lacp->actor_port_priority;
member->partner_port_id = lacp->actor_port_id;
member->partner_state = lacp->actor_state;
member->timeout = 0;
member->actor_state &= ~(LACP_STATE_FLAG_DEFAULTED|LACP_STATE_FLAG_EXPIRED);
member->stats.lacp_rx++;
if(member->partner_state != lacp->actor_state ||
member->partner_system_priority != lacp->actor_system_priority) {
memcpy(member->partner_system_id, lacp->actor_system_id, ETH_ADDR_LEN);
member->partner_system_priority = lacp->actor_system_priority;
member->partner_key = lacp->actor_key;
member->partner_port_priority = lacp->actor_port_priority;
member->partner_port_id = lacp->actor_port_id;
member->partner_state = lacp->actor_state;
bbl_lag_select(member->lag);
}
}
return;
}
code/bngblaster/src/bbl_lag.h
+14 -6
View File
@@ -10,23 +10,30 @@
#ifndef __BBL_LAG_H__
#define __BBL_LAG_H__
#define LAG_MEMBER_ACTIVE_MAX 16
typedef struct bbl_lag_
{
interface_state_t state;
uint8_t id;
char *interface;
bbl_interface_s *interface;
bbl_lag_config_s *config;
CIRCLEQ_HEAD(lag_interface_, bbl_interface_ ) lag_interface_qhead; /* list of interfaces */
uint8_t active_max;
uint8_t active_count;
bbl_lag_member_s *active_list[LAG_MEMBER_ACTIVE_MAX];
CIRCLEQ_ENTRY(bbl_lag_) lag_qnode;
CIRCLEQ_HEAD(lag_member_, bbl_lag_member_ ) lag_member_qhead; /* list of member interfaces */
} bbl_lag_s;
typedef struct bbl_lag_member_
{
bbl_lag_s *lag;
bbl_interface_s *interface;
lacp_state_t lacp_state;
interface_state_t state;
bool primary;
bool periodic_fast;
struct timer_ *lacp_timer;
uint8_t timeout;
@@ -45,6 +52,7 @@ typedef struct bbl_lag_member_
uint16_t partner_port_id;
uint8_t partner_state;
CIRCLEQ_ENTRY(bbl_lag_member_) lag_member_qnode;
struct {
uint32_t lacp_rx;
uint32_t lacp_tx;
code/bngblaster/src/bbl_network.c
+10 -19
View File
@@ -31,18 +31,6 @@ bbl_network_interface_rate_job(timer_s *timer) {
bbl_compute_avg_rate(&interface->stats.rate_session_ipv6pd_rx, interface->stats.session_ipv6pd_rx);
}
static bool
bbl_network_interfaces_duplicate(bbl_interface_s *interface, uint16_t vlan) {
bbl_network_interface_s *network_interface = interface->network;
while(network_interface) {
if(network_interface->vlan == vlan) {
return true;
}
network_interface = network_interface->next;
}
return false;
}
/**
* bbl_network_interfaces_add
*/
@@ -66,29 +54,32 @@ bbl_network_interfaces_add()
snprintf(ifname, sizeof(ifname), "%s",
network_config->interface);
}
interface = bbl_interface_get(network_config->interface);
if (!interface) {
LOG(ERROR, "Failed to add network interface %s (interface not found)\n", ifname);
return false;
}
if(bbl_network_interface_get(ifname)) {
LOG(ERROR, "Failed to add network interface %s (duplicate)\n", ifname);
return false;
}
if(interface->access && network_config->vlan == 0) {
LOG(ERROR, "Failed to add network interface %s (untagged not allowed on access interfaces)\n", ifname);
return false;
}
if(bbl_network_interfaces_duplicate(interface, network_config->vlan)) {
LOG(ERROR, "Failed to add network interface %s (duplicate)\n", ifname);
return false;
}
network_interface = calloc(1, sizeof(bbl_network_interface_s));
network_interface->next = interface->network;
interface->network = network_interface;
network_interface->interface = interface;
network_config->network_interface = network_interface;
network_interface->name = strdup(ifname);
CIRCLEQ_INSERT_TAIL(&g_ctx->network_interface_qhead, network_interface, network_interface_qnode);
/* Init interface */
network_interface->name = strdup(ifname);
network_interface->interface = interface;
/* Init TXQ */
network_interface->txq = calloc(1, sizeof(bbl_txq_s));
bbl_txq_init(network_interface->txq, BBL_TXQ_DEFAULT_SIZE);
code/bngblaster/src/bbl_network.h
+8 -2
View File
@@ -12,9 +12,15 @@
typedef struct bbl_network_interface_
{
char *name;
bbl_interface_s *interface;
char *name; /* interface name */
uint32_t ifindex; /* interface index */
/* parent */
bbl_interface_s *interface;
/* next network interface with same parent */
struct bbl_network_interface_ *next;
bbl_txq_s *txq;
uint16_t vlan;
code/bngblaster/src/bbl_protocols.c
+10 -10
View File
@@ -2050,23 +2050,23 @@ decode_lacp(uint8_t *buf, uint16_t len,
if(tlv_len != 20) {
return DECODE_ERROR;
}
lacp->actor_system_priority = be16toh(*(uint16_t*)buf+2);
lacp->actor_system_priority = be16toh(*(uint16_t*)(buf+2));
lacp->actor_system_id = buf+4;
lacp->actor_key = be16toh(*(uint16_t*)buf+10);
lacp->actor_port_priority = be16toh(*(uint16_t*)buf+12);
lacp->actor_port_id = be16toh(*(uint16_t*)buf+14);
lacp->actor_state = *buf+16;
lacp->actor_key = be16toh(*(uint16_t*)(buf+10));
lacp->actor_port_priority = be16toh(*(uint16_t*)(buf+12));
lacp->actor_port_id = be16toh(*(uint16_t*)(buf+14));
lacp->actor_state = *(buf+16);
break;
case LACP_TLV_PARTNER_INFORMATION:
if(tlv_len != 20) {
return DECODE_ERROR;
}
lacp->partner_system_priority = be16toh(*(uint16_t*)buf+2);
lacp->partner_system_priority = be16toh(*(uint16_t*)(buf+2));
lacp->partner_system_id = buf+4;
lacp->partner_key = be16toh(*(uint16_t*)buf+10);
lacp->partner_port_priority = be16toh(*(uint16_t*)buf+12);
lacp->partner_port_id = be16toh(*(uint16_t*)buf+14);
lacp->partner_state = *buf+16;
lacp->partner_key = be16toh(*(uint16_t*)(buf+10));
lacp->partner_port_priority = be16toh(*(uint16_t*)(buf+12));
lacp->partner_port_id = be16toh(*(uint16_t*)(buf+14));
lacp->partner_state = *(buf+16);
break;
default:
break;
code/bngblaster/src/bbl_rx.c
+5
View File
@@ -111,6 +111,11 @@ bbl_rx_handler(bbl_interface_s *interface,
break;
}
if(interface->type == LAG_MEMBER_INTERFACE) {
bbl_rx_handler(interface->lag->interface, eth);
return;
}
while(network_interface) {
if(network_interface->vlan == eth->vlan_outer) {
if(!bbl_rx_stream_network(network_interface, eth)) {
code/bngblaster/src/bbl_session.c
+11 -8
View File
@@ -16,7 +16,8 @@ extern volatile bool g_teardown;
extern volatile bool g_monkey;
const char *
session_state_string(uint32_t state) {
session_state_string(uint32_t state)
{
switch(state) {
case BBL_IDLE: return "Idle";
case BBL_IPOE_SETUP: return "IPoE Setup";
@@ -34,7 +35,8 @@ session_state_string(uint32_t state) {
}
const char *
ppp_state_string(uint32_t state) {
ppp_state_string(uint32_t state)
{
switch(state) {
case BBL_PPP_CLOSED: return "Closed";
case BBL_PPP_INIT: return "Init";
@@ -47,7 +49,8 @@ ppp_state_string(uint32_t state) {
}
const char *
dhcp_state_string(uint32_t state) {
dhcp_state_string(uint32_t state)
{
switch(state) {
case BBL_DHCP_INIT: return "Init";
case BBL_DHCP_SELECTING: return "Selecting";
@@ -685,7 +688,7 @@ bbl_sessions_init()
session->access_type = access_config->access_type;
session->access_interface = access_config->access_interface;
session->network_interface = bbl_network_interface_get(access_config->network_interface);
session->vlan_key.ifindex = access_config->access_interface->interface->ifindex;
session->vlan_key.ifindex = access_config->access_interface->ifindex;
session->vlan_key.outer_vlan_id= access_config->access_outer_vlan;
session->vlan_key.inner_vlan_id = access_config->access_inner_vlan;
session->access_third_vlan = access_config->access_third_vlan;
@@ -847,7 +850,7 @@ bbl_sessions_init()
}
LOG(DEBUG, "Session %u created (%s.%u:%u)\n", i,
access_config->access_interface->interface->name,
access_config->access_interface->name,
access_config->access_outer_vlan,
access_config->access_inner_vlan);
@@ -1045,7 +1048,7 @@ bbl_session_json(bbl_session_s *session)
}
if(session->a10nsp_session) {
a10nsp_session = json_pack("{ss si sb sb ss* ss* ss* ss* ss* ss* si si}",
"interface", session->a10nsp_session->a10nsp_interface->interface->name,
"interface", session->a10nsp_session->a10nsp_interface->name,
"s-vlan", session->a10nsp_session->s_vlan,
"qinq-send", session->a10nsp_session->a10nsp_interface->qinq,
"qinq-received", session->a10nsp_session->qinq_received,
@@ -1064,7 +1067,7 @@ bbl_session_json(bbl_session_s *session)
"type", "pppoe",
"session-id", session->session_id,
"session-state", session_state_string(session->session_state),
"interface", session->access_interface->interface->name,
"interface", session->access_interface->name,
"outer-vlan", session->vlan_key.outer_vlan_id,
"inner-vlan", session->vlan_key.inner_vlan_id,
"mac", format_mac_address(session->client_mac),
@@ -1112,7 +1115,7 @@ bbl_session_json(bbl_session_s *session)
"type", "ipoe",
"session-id", session->session_id,
"session-state", session_state_string(session->session_state),
"interface", session->access_interface->interface->name,
"interface", session->access_interface->name,
"outer-vlan", session->vlan_key.outer_vlan_id,
"inner-vlan", session->vlan_key.inner_vlan_id,
"mac", format_mac_address(session->client_mac),
code/bngblaster/src/bbl_stream.c
+282 -277
View File
@@ -40,38 +40,6 @@ bbl_stream_delay(bbl_stream_s *stream, struct timespec *rx_timestamp, struct tim
}
}
static bool
bbl_stream_can_send(bbl_stream_s *stream)
{
if(g_init_phase) {
return false;
}
if(stream->reset) {
stream->reset = false;
stream->flow_seq = 1;
goto FREE;
}
if(stream->interface->state != INTERFACE_UP) {
goto FREE;
}
if(*(stream->endpoint) == ENDPOINT_ACTIVE) {
return true;
}
FREE:
/* Free packet if not ready to send */
if(stream->buf) {
free(stream->buf);
stream->buf = NULL;
stream->tx_len = 0;
}
stream->send_window_packets = 0;
return false;
}
static bool
bbl_stream_build_access_pppoe_packet(bbl_stream_s *stream)
{
@@ -197,10 +165,10 @@ bbl_stream_build_access_pppoe_packet(bbl_stream_s *stream)
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -315,10 +283,10 @@ bbl_stream_build_a10nsp_pppoe_packet(bbl_stream_s *stream)
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -438,10 +406,10 @@ bbl_stream_build_a10nsp_ipoe_packet(bbl_stream_s *stream)
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -570,10 +538,10 @@ bbl_stream_build_access_ipoe_packet(bbl_stream_s *stream)
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -712,10 +680,10 @@ bbl_stream_build_network_packet(bbl_stream_s *stream)
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -793,10 +761,10 @@ bbl_stream_build_l2tp_packet(bbl_stream_s *stream)
}
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_buf = malloc(buf_len);
if(encode_ethernet(stream->tx_buf, &stream->tx_len, &eth) != PROTOCOL_SUCCESS) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
return false;
}
@@ -849,63 +817,6 @@ bbl_stream_build_packet(bbl_stream_s *stream)
return false;
}
static uint64_t
bbl_stream_send_window(bbl_stream_s *stream, struct timespec *now) {
uint64_t packets = 0;
uint64_t packets_expected;
struct timespec time_elapsed = {0};
/** Enforce optional stream traffic start delay ... */
if(stream->config->start_delay && stream->packets_tx == 0) {
if(stream->wait) {
timespec_sub(&time_elapsed, now, &stream->wait_start);
if(time_elapsed.tv_sec < stream->config->start_delay) {
/** Still waiting ... */
return 0;
}
/** Stop wait window ... */
} else {
/** Start wait window ... */
stream->wait = true;
stream->wait_start.tv_sec = now->tv_sec;
stream->wait_start.tv_nsec = now->tv_nsec;
return 0;
}
}
if(stream->send_window_packets) {
timespec_sub(&time_elapsed, now, &stream->send_window_start);
packets_expected = time_elapsed.tv_sec * stream->config->pps;
packets_expected += stream->config->pps * ((double)time_elapsed.tv_nsec / 1000000000.0);
if(packets_expected > stream->send_window_packets) {
packets = packets_expected - stream->send_window_packets;
}
if(packets > g_ctx->config.stream_max_ppi) {
packets = g_ctx->config.stream_max_ppi;
}
} else {
/* Open new send window */
stream->send_window_start.tv_sec = now->tv_sec;
stream->send_window_start.tv_nsec = now->tv_nsec;
packets = 1;
}
/** Enforce optional stream packet limit ... */
if(stream->config->max_packets &&
stream->packets_tx + packets > stream->config->max_packets) {
if(stream->packets_tx < stream->config->max_packets) {
packets = stream->config->max_packets - stream->packets_tx;
} else {
packets = 0;
}
}
return packets;
}
static void
bbl_stream_tx_stats(bbl_stream_s *stream, uint64_t packets, uint64_t bytes)
{
@@ -1227,21 +1138,102 @@ bbl_stream_final()
dict_itor_free(itor);
}
void
bbl_stream_group_ctrl_job(timer_s *timer)
static bool
bbl_stream_can_send(bbl_stream_s *stream)
{
bbl_stream_group_s *group = timer->data;
bbl_stream_s *stream = group->stream_head;
while(stream) {
bbl_stream_ctrl(stream);
stream = stream->group_next;
if(g_init_phase) {
return false;
}
if(stream->reset) {
stream->reset = false;
stream->flow_seq = 1;
goto FREE;
}
if(stream->tx_interface->state != INTERFACE_UP) {
goto FREE;
}
if(*(stream->endpoint) == ENDPOINT_ACTIVE) {
return true;
}
FREE:
/* Free packet if not ready to send */
if(stream->tx_buf) {
free(stream->tx_buf);
stream->tx_buf = NULL;
stream->tx_len = 0;
}
stream->send_window_packets = 0;
return false;
}
bool
bbl_stream_tx(bbl_stream_group_s *group, bbl_stream_s *stream)
static uint64_t
bbl_stream_send_window(bbl_stream_s *stream, struct timespec *now) {
uint64_t packets = 1;
uint64_t packets_expected;
struct timespec time_elapsed = {0};
/** Enforce optional stream traffic start delay ... */
if(stream->config->start_delay && stream->packets_tx == 0) {
if(stream->wait) {
timespec_sub(&time_elapsed, now, &stream->wait_start);
if(time_elapsed.tv_sec < stream->config->start_delay) {
/** Still waiting ... */
return 0;
}
/** Stop wait window ... */
} else {
/** Start wait window ... */
stream->wait = true;
stream->wait_start.tv_sec = now->tv_sec;
stream->wait_start.tv_nsec = now->tv_nsec;
return 0;
}
}
if(stream->send_window_packets) {
timespec_sub(&time_elapsed, now, &stream->send_window_start);
packets_expected = time_elapsed.tv_sec * stream->config->pps;
packets_expected += stream->config->pps * ((double)time_elapsed.tv_nsec / 1000000000.0);
if(packets_expected > stream->send_window_packets) {
packets = packets_expected - stream->send_window_packets;
}
if(packets > g_ctx->config.stream_max_ppi) {
packets = g_ctx->config.stream_max_ppi;
}
} else {
/* Open new send window */
stream->send_window_start.tv_sec = now->tv_sec;
stream->send_window_start.tv_nsec = now->tv_nsec;
packets = 1;
}
/** Enforce optional stream packet limit ... */
if(stream->config->max_packets &&
stream->packets_tx + packets > stream->config->max_packets) {
if(stream->packets_tx < stream->config->max_packets) {
packets = stream->config->max_packets - stream->packets_tx;
} else {
packets = 0;
}
}
return packets;
}
static bool
bbl_stream_tx(bbl_stream_s *stream)
{
bbl_session_s *session = stream->session;
io_handle_s *io = stream->io;
struct timespec now;
uint64_t packets = 1;
uint64_t send = 0;
@@ -1250,7 +1242,7 @@ bbl_stream_tx(bbl_stream_group_s *group, bbl_stream_s *stream)
return true;
}
if(!stream->buf) {
if(!stream->tx_buf) {
if(!bbl_stream_build_packet(stream)) {
LOG(ERROR, "Failed to build packet for stream %s\n", stream->config->name);
return true;
@@ -1274,14 +1266,15 @@ bbl_stream_tx(bbl_stream_group_s *group, bbl_stream_s *stream)
}
}
packets = bbl_stream_send_window(stream, &group->tx_timestamp);
clock_gettime(CLOCK_MONOTONIC, &now);
packets = bbl_stream_send_window(stream, &now);
/* Update BBL header fields */
*(uint32_t*)(stream->buf + (stream->tx_len - 8)) = group->tx_timestamp.tv_sec;
*(uint32_t*)(stream->buf + (stream->tx_len - 4)) = group->tx_timestamp.tv_nsec;
*(uint32_t*)(stream->tx_buf + (stream->tx_len - 8)) = now.tv_sec;
*(uint32_t*)(stream->tx_buf + (stream->tx_len - 4)) = now.tv_nsec;
while(packets) {
*(uint64_t*)(stream->buf + (stream->tx_len - 16)) = stream->flow_seq;
*(uint64_t*)(stream->tx_buf + (stream->tx_len - 16)) = stream->flow_seq;
/* Send packet */
if(!io_send(group->io, stream->buf, stream->tx_len)) {
if(!io_send(io, stream->tx_buf, stream->tx_len)) {
return false;
}
stream->send_window_packets++;
@@ -1294,103 +1287,111 @@ bbl_stream_tx(bbl_stream_group_s *group, bbl_stream_s *stream)
}
void
bbl_stream_group_tx_job(timer_s *timer)
bbl_stream_tx_job(timer_s *timer)
{
bbl_stream_tx(timer->data);
}
void
bbl_stream_lag_tx_job(timer_s *timer)
{
bbl_stream_s *stream = timer->data;
bbl_lag_s *lag = stream->tx_interface->lag;
uint8_t key = stream->flow_id % lag->active_count;
stream->io = lag->active_list[key]->interface->io.tx;
bbl_stream_tx(stream);
}
void
bbl_stream_group_job(timer_s *timer)
{
bbl_stream_group_s *group = timer->data;
bbl_stream_s *stream;
clock_gettime(CLOCK_MONOTONIC, &group->tx_timestamp);
stream = group->stream_pos;
bbl_stream_s *stream = group->head;
while(stream) {
if(!bbl_stream_tx(group, stream)) {
group->stream_pos = stream;
return;
}
bbl_stream_ctrl(stream);
stream = stream->group_next;
if(!stream) {
stream = group->stream_head;
}
if(stream == group->stream_pos) {
break;
}
}
}
bbl_stream_group_s *
bbl_stream_group_init()
{
bbl_stream_group_s *group = calloc(1, sizeof(bbl_stream_group_s));
timer_add_periodic(&g_ctx->timer_root, &group->timer, "Stream CTRL",
1, 0, group, &bbl_stream_group_job);
return group;
}
static void
bbl_stream_group(io_handle_s *io, bbl_stream_s *stream)
bbl_stream_add_group(bbl_stream_s *stream)
{
bbl_stream_group_s *group = io->stream_group;
io_thread_s *thread = io->thread;
time_t timer_sec;
long timer_nsec;
while(group) {
if((group->count < g_ctx->config.stream_max_per_group) &&
(group->tx_interval == stream->tx_interval)) {
break;
}
group = group->next;
bbl_stream_group_s *group = NULL;
if(!g_ctx->stream_groups) {
group = bbl_stream_group_init();
g_ctx->stream_groups = group;
} else if(g_ctx->stream_groups->count >= 64) {
group = bbl_stream_group_init();
group->next = g_ctx->stream_groups;
g_ctx->stream_groups = group;
} else {
group = g_ctx->stream_groups;
}
if(!group) {
/* Create new group */
group = calloc(1, sizeof(bbl_stream_group_s));
group->next = io->stream_group;
io->stream_group = group;
group->io = io;
group->tx_interval = stream->tx_interval;
timer_sec = group->tx_interval / 1000000000;
timer_nsec = group->tx_interval % 1000000000;
if(thread) {
timer_add_periodic(&thread->timer.root, &group->timer_tx, "Stream Group TX",
timer_sec, timer_nsec, group, &bbl_stream_group_tx_job);
} else {
timer_add_periodic(&g_ctx->timer_root, &group->timer_tx, "Stream Group TX",
timer_sec, timer_nsec, group, &bbl_stream_group_tx_job);
}
timer_add_periodic(&g_ctx->timer_root, &group->timer_ctrl, "Stream Group CTRL",
1, 0, group, &bbl_stream_group_ctrl_job);
}
/* Add stream to group */
stream->group = group;
stream->group_next = group->stream_head;
group->stream_head = stream;
group->stream_pos = stream;
stream->group_next = group->head;
group->head = stream;
group->count++;
}
static bool
bbl_stream_add_jobs(bbl_stream_s *stream)
{
io_handle_s *io = stream->interface->io.tx;
static void
bbl_stream_select_io(bbl_stream_s *stream)
{
io_handle_s *io = stream->tx_interface->io.tx;
io_thread_s *thread = io->thread;
if(thread) {
stream->threaded = true;
while(io && io->thread) {
if(io->thread->pps_reserved < thread->pps_reserved) {
thread = io->thread;
}
io = io->next;
}
thread->pps_reserved += stream->config->pps;
stream->thread = thread;
io = thread->io;
}
stream->io = io;
}
bbl_stream_group(io, stream);
return true;
static void
bbl_stream_add(bbl_stream_s *stream)
{
time_t timer_sec;
long timer_nsec;
bbl_stream_add_group(stream);
timer_sec = stream->tx_interval / 1000000000;
timer_nsec = stream->tx_interval % 1000000000;
if(stream->tx_interface->type == LAG_INTERFACE) {
timer_add_periodic(&g_ctx->timer_root, &stream->tx_timer, "Stream TX",
timer_sec, timer_nsec, stream, &bbl_stream_lag_tx_job);
} else {
bbl_stream_select_io(stream);
if(stream->io->thread) {
timer_add_periodic(&stream->io->thread->timer.root, &stream->tx_timer, "Stream TX",
timer_sec, timer_nsec, stream, &bbl_stream_tx_job);
} else {
timer_add_periodic(&g_ctx->timer_root, &stream->tx_timer, "Stream TX",
timer_sec, timer_nsec, stream, &bbl_stream_tx_job);
}
}
}
static bool
bbl_stream_add(bbl_stream_config_s *config, bbl_session_s *session)
bbl_stream_session_add(bbl_stream_config_s *config, bbl_session_s *session)
{
bbl_interface_s *interface = NULL;
bbl_access_interface_s *access_interface = NULL;
bbl_network_interface_s *network_interface = NULL;
bbl_a10nsp_interface_s *a10nsp_interface = NULL;
bbl_stream_s *stream = NULL;
@@ -1399,6 +1400,10 @@ bbl_stream_add(bbl_stream_config_s *config, bbl_session_s *session)
uint64_t tx_interval = 0;
assert(config);
assert(session);
access_interface = session->access_interface;
/* *
* The corresponding network/a01nsp interfaces will be selected
* in the following order:
@@ -1421,17 +1426,7 @@ bbl_stream_add(bbl_stream_config_s *config, bbl_session_s *session)
}
}
if(network_interface) {
interface = network_interface->interface;
} else if(a10nsp_interface) {
interface = a10nsp_interface->interface;
} else {
LOG_NOARG(ERROR, "Failed to add stream because of missing network/a01nsp interface\n");
return false;
}
tx_interval = SEC / config->pps;
tx_interval = tx_interval / g_ctx->config.stream_tx_precision;
if(config->direction & BBL_DIRECTION_UP) {
stream = calloc(1, sizeof(bbl_stream_s));
stream->endpoint = &g_endpoint;
@@ -1441,45 +1436,43 @@ bbl_stream_add(bbl_stream_config_s *config, bbl_session_s *session)
stream->type = BBL_TYPE_UNICAST;
stream->sub_type = config->type;
stream->direction = BBL_DIRECTION_UP;
if(session) {
stream->session = session;
stream->interface = session->access_interface->interface;
stream->access_interface = session->access_interface;
switch(stream->sub_type) {
case BBL_SUB_TYPE_IPV4:
stream->endpoint = &(session->endpoint.ipv4);
break;
case BBL_SUB_TYPE_IPV6:
stream->endpoint = &(session->endpoint.ipv6);
break;
case BBL_SUB_TYPE_IPV6PD:
stream->endpoint = &(session->endpoint.ipv6pd);
break;
default:
break;
}
}
stream->tx_interval = tx_interval;
stream->session_traffic = config->session_traffic;
stream->session = session;
switch(stream->sub_type) {
case BBL_SUB_TYPE_IPV4:
stream->endpoint = &(session->endpoint.ipv4);
break;
case BBL_SUB_TYPE_IPV6:
stream->endpoint = &(session->endpoint.ipv6);
break;
case BBL_SUB_TYPE_IPV6PD:
stream->endpoint = &(session->endpoint.ipv6pd);
break;
default:
break;
}
stream->access_interface = access_interface;
stream->tx_interface = access_interface->interface;
stream->tx_interval = tx_interval;
result = dict_insert(g_ctx->stream_flow_dict, &stream->flow_id);
if(!result.inserted) {
LOG(ERROR, "Failed to insert stream %s\n", config->name);
LOG(ERROR, "Failed to insert stream %s (upstream)\n", config->name);
free(stream);
return false;
}
*result.datum_ptr = stream;
stream->session_next = session->streams.head;
session->streams.head = stream;
bbl_stream_add_jobs(stream);
bbl_stream_add(stream);
if(stream->session_traffic) {
g_ctx->stats.session_traffic_flows++;
session->session_traffic.flows++;
LOG(DEBUG, "Session traffic stream %s added to %s (upstream) with %lf PPS\n",
stream->interface->name, config->name, config->pps);
LOG(DEBUG, "Session traffic stream %s (upstream) added to %s (access) with %lf PPS\n",
access_interface->name, config->name, config->pps);
} else {
g_ctx->stats.stream_traffic_flows++;
LOG(DEBUG, "Traffic stream %s added to %s (upstream) with %lf PPS\n",
stream->interface->name, config->name, config->pps);
LOG(DEBUG, "Traffic stream %s (upstream) added to %s (access) with %lf PPS\n",
access_interface->name, config->name, config->pps);
}
}
if(config->direction & BBL_DIRECTION_DOWN) {
@@ -1491,46 +1484,62 @@ bbl_stream_add(bbl_stream_config_s *config, bbl_session_s *session)
stream->type = BBL_TYPE_UNICAST;
stream->sub_type = config->type;
stream->direction = BBL_DIRECTION_DOWN;
stream->interface = interface;
stream->network_interface = network_interface;
stream->a10nsp_interface = a10nsp_interface;
if(session) {
stream->session = session;
switch(stream->sub_type) {
case BBL_SUB_TYPE_IPV4:
stream->endpoint = &session->endpoint.ipv4;
break;
case BBL_SUB_TYPE_IPV6:
stream->endpoint = &session->endpoint.ipv6;
break;
case BBL_SUB_TYPE_IPV6PD:
stream->endpoint = &session->endpoint.ipv6pd;
break;
default:
break;
}
stream->session = session;
switch(stream->sub_type) {
case BBL_SUB_TYPE_IPV4:
stream->endpoint = &session->endpoint.ipv4;
break;
case BBL_SUB_TYPE_IPV6:
stream->endpoint = &session->endpoint.ipv6;
break;
case BBL_SUB_TYPE_IPV6PD:
stream->endpoint = &session->endpoint.ipv6pd;
break;
default:
break;
}
stream->tx_interval = tx_interval;
stream->session_traffic = config->session_traffic;
result = dict_insert(g_ctx->stream_flow_dict, &stream->flow_id);
if(!result.inserted) {
LOG(ERROR, "Failed to insert stream %s\n", config->name);
LOG(ERROR, "Failed to insert stream %s (downstream)\n", config->name);
free(stream);
return false;
}
*result.datum_ptr = stream;
stream->session_next = session->streams.head;
session->streams.head = stream;
bbl_stream_add_jobs(stream);
if(stream->session_traffic) {
g_ctx->stats.session_traffic_flows++;
session->session_traffic.flows++;
LOG(DEBUG, "Session traffic stream %s added to %s (downstream) with %lf PPS\n",
stream->interface->name, config->name, config->pps);
if(network_interface) {
stream->network_interface = network_interface;
stream->tx_interface = network_interface->interface;
bbl_stream_add(stream);
if(stream->session_traffic) {
g_ctx->stats.session_traffic_flows++;
session->session_traffic.flows++;
LOG(DEBUG, "Session traffic stream %s (downstream) added to %s (network) with %lf PPS\n",
network_interface->name, config->name, config->pps);
} else {
g_ctx->stats.stream_traffic_flows++;
LOG(DEBUG, "Traffic stream %s (downstream) added to %s (network) with %lf PPS\n",
network_interface->name, config->name, config->pps);
}
} else if(a10nsp_interface) {
stream->a10nsp_interface = a10nsp_interface;
stream->tx_interface = a10nsp_interface->interface;
bbl_stream_add(stream);
if(stream->session_traffic) {
g_ctx->stats.session_traffic_flows++;
session->session_traffic.flows++;
LOG(DEBUG, "Session traffic stream %s (downstream) added to %s (a10nsp) with %lf PPS\n",
network_interface->name, config->name, config->pps);
} else {
g_ctx->stats.stream_traffic_flows++;
LOG(DEBUG, "Traffic stream %s (downstream) added to %s (a10nsp) with %lf PPS\n",
network_interface->name, config->name, config->pps);
}
} else {
g_ctx->stats.stream_traffic_flows++;
LOG(DEBUG, "Traffic stream %s added to %s (downstream) with %lf PPS\n",
stream->interface->name, config->name, config->pps);
LOG(ERROR, "Failed to add stream %s (downstream) because of missing interface\n", config->name);
return false;
}
}
return true;
@@ -1543,37 +1552,37 @@ bbl_stream_session_init(bbl_session_s *session)
/** Add session traffic ... */
if(g_ctx->config.stream_config_session_ipv4_up && session->endpoint.ipv4) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv4_up, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv4_up, session)) {
return false;
}
session->session_traffic.ipv4_up = session->streams.head;
}
if(g_ctx->config.stream_config_session_ipv4_down && session->endpoint.ipv4) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv4_down, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv4_down, session)) {
return false;
}
session->session_traffic.ipv4_down = session->streams.head;
}
if(g_ctx->config.stream_config_session_ipv6_up && session->endpoint.ipv6) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv6_up, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv6_up, session)) {
return false;
}
session->session_traffic.ipv6_up = session->streams.head;
}
if(g_ctx->config.stream_config_session_ipv6_down && session->endpoint.ipv6) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv6_down, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv6_down, session)) {
return false;
}
session->session_traffic.ipv6_down = session->streams.head;
}
if(g_ctx->config.stream_config_session_ipv6pd_up && session->endpoint.ipv6pd) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv6pd_up, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv6pd_up, session)) {
return false;
}
session->session_traffic.ipv6pd_up = session->streams.head;
}
if(g_ctx->config.stream_config_session_ipv6pd_down && session->endpoint.ipv6pd) {
if(!bbl_stream_add(g_ctx->config.stream_config_session_ipv6pd_down, session)) {
if(!bbl_stream_session_add(g_ctx->config.stream_config_session_ipv6pd_down, session)) {
return false;
}
session->session_traffic.ipv6pd_down = session->streams.head;
@@ -1584,7 +1593,7 @@ bbl_stream_session_init(bbl_session_s *session)
config = g_ctx->config.stream_config;
while(config) {
if(config->stream_group_id == session->streams.group_id) {
if(!bbl_stream_add(config, session)) {
if(!bbl_stream_session_add(config, session)) {
return false;
}
}
@@ -1617,13 +1626,11 @@ bbl_stream_init() {
if(config->stream_group_id == 0) {
network_interface = bbl_network_interface_get(config->network_interface);
if(!network_interface) {
LOG_NOARG(ERROR, "Failed to add RAW stream because of missing network interface\n");
LOG(ERROR, "Failed to add RAW stream %s because of missing network interface\n", config->name);
return false;
}
tx_interval = SEC / config->pps;
tx_interval = tx_interval / g_ctx->config.stream_tx_precision;
if(config->direction & BBL_DIRECTION_DOWN) {
stream = calloc(1, sizeof(bbl_stream_s));
stream->endpoint = &g_endpoint;
@@ -1639,8 +1646,8 @@ bbl_stream_init() {
}
}
stream->direction = BBL_DIRECTION_DOWN;
stream->interface = network_interface->interface;
stream->network_interface = network_interface;
stream->tx_interface = network_interface->interface;
stream->tx_interval = tx_interval;
result = dict_insert(g_ctx->stream_flow_dict, &stream->flow_id);
if(!result.inserted) {
@@ -1649,7 +1656,7 @@ bbl_stream_init() {
return false;
}
*result.datum_ptr = stream;
bbl_stream_add_jobs(stream);
bbl_stream_add(stream);
if(stream->type == BBL_TYPE_MULTICAST) {
LOG(DEBUG, "RAW multicast traffic stream %s added to %s with %lf PPS\n",
config->name, network_interface->name, config->pps);
@@ -1667,13 +1674,11 @@ bbl_stream_init() {
if(g_ctx->config.send_multicast_traffic && g_ctx->config.igmp_group_count) {
network_interface = bbl_network_interface_get(g_ctx->config.multicast_traffic_network_interface);
if(!network_interface) {
LOG_NOARG(ERROR, "Failed to add multicast streams because of missing network interface\n");
LOG_NOARG(ERROR, "Failed to add autogenerated multicast streams because of missing network interface\n");
return false;
}
tx_interval = SEC / g_ctx->config.multicast_traffic_pps;
tx_interval = tx_interval / g_ctx->config.stream_tx_precision;
for(i = 0; i < g_ctx->config.igmp_group_count; i++) {
group = be32toh(g_ctx->config.igmp_group) + i * be32toh(g_ctx->config.igmp_group_iter);
@@ -1707,8 +1712,8 @@ bbl_stream_init() {
stream->type = BBL_TYPE_MULTICAST;
stream->sub_type = config->type;
stream->direction = BBL_DIRECTION_DOWN;
stream->interface = network_interface->interface;
stream->network_interface = network_interface;
stream->tx_interface = network_interface->interface;
stream->tx_interval = tx_interval;
result = dict_insert(g_ctx->stream_flow_dict, &stream->flow_id);
if(!result.inserted) {
@@ -1717,7 +1722,7 @@ bbl_stream_init() {
return false;
}
*result.datum_ptr = stream;
bbl_stream_add_jobs(stream);
bbl_stream_add(stream);
LOG(DEBUG, "Autogenerated multicast traffic stream added to %s with %lf PPS\n",
network_interface->name, config->pps);
}
@@ -1969,13 +1974,13 @@ bbl_stream_json(bbl_stream_s *stream)
}
if(stream->access_interface) {
access_interface_name = stream->access_interface->interface->name;
access_interface_name = stream->access_interface->name;
}
if(stream->network_interface) {
network_interface_name = stream->network_interface->name;
}
if(stream->a10nsp_interface) {
a10nsp_interface_name = stream->a10nsp_interface->interface->name;
a10nsp_interface_name = stream->a10nsp_interface->name;
}
if(stream->type == BBL_TYPE_UNICAST) {
code/bngblaster/src/bbl_stream.h
+9 -19
View File
@@ -61,19 +61,10 @@ typedef struct bbl_stream_config_
typedef struct bbl_stream_group_
{
uint32_t count;
bbl_stream_s *head;
struct timer_ *timer;
bbl_stream_group_s *next;
io_handle_s *io;
bbl_stream_s *stream_head;
bbl_stream_s *stream_pos;
struct timer_ *timer_tx;
struct timespec tx_timestamp;
uint64_t tx_interval; /* TX interval in nsec */
uint32_t count; /* stream count */
char _pad0 __attribute__((__aligned__(CACHE_LINE_SIZE))); /* empty cache line */
struct timer_ *timer_ctrl;
} bbl_stream_group_s;
typedef struct bbl_stream_
@@ -94,17 +85,19 @@ typedef struct bbl_stream_
bbl_stream_s *session_next; /* Next stream of same session */
endpoint_state_t *endpoint;
io_thread_s *thread;
io_handle_s *io;
bbl_interface_s *interface;
bbl_access_interface_s *access_interface;
bbl_network_interface_s *network_interface;
bbl_a10nsp_interface_s *a10nsp_interface;
uint8_t *buf;
uint16_t tx_len;
struct timer_ *tx_timer;
bbl_interface_s *tx_interface; /* TX interface */
uint8_t *tx_buf; /* TX buffer */
uint16_t tx_len; /* TX length */
uint64_t tx_interval; /* TX interval in nsec */
bool threaded;
bool session_traffic;
bool verified;
bool wait;
@@ -160,9 +153,6 @@ typedef struct bbl_stream_
} bbl_stream_s;
void
bbl_stream_tx_job(timer_s *timer);
bool
bbl_stream_session_init(bbl_session_s *session);
code/bngblaster/src/bbl_tx.c
+14 -2
View File
@@ -1416,7 +1416,10 @@ bbl_tx_encode_network_packet(bbl_network_interface_s *interface, uint8_t *buf, u
} else {
interface->send_requests = 0;
}
if(result == PROTOCOL_SUCCESS) {
interface->stats.packets_tx++;
interface->stats.bytes_tx += *len;
}
return result;
}
@@ -1432,11 +1435,12 @@ bbl_tx_encode_interface_packet(bbl_interface_s *interface,
if(interface->send_requests & BBL_SEND_LACP) {
interface->send_requests &= ~BBL_SEND_LACP;
member = interface->lag_member;
member->stats.lacp_tx++;
eth.src = interface->mac;
eth.dst = (uint8_t*)slow_mac;
eth.type = ETH_TYPE_LACP;
eth.next = &lacp;
member = interface->lag;
lacp.actor_system_id = member->actor_system_id;
lacp.actor_system_priority = member->actor_system_priority;
lacp.actor_key = member->actor_key;
@@ -1486,6 +1490,14 @@ bbl_tx(bbl_interface_s *interface, uint8_t *buf, uint16_t *len)
return EMPTY;
}
if(interface->type == LAG_MEMBER_INTERFACE) {
if(interface->lag_member->primary) {
return bbl_tx(interface->lag->interface, buf, len);
} else {
return EMPTY;
}
}
if(interface->access) {
access_interface = interface->access;
if(!bbl_txq_is_empty(access_interface->txq)) {
code/bngblaster/src/io/io_def.h
-1
View File
@@ -56,7 +56,6 @@ typedef struct io_handle_ {
io_thread_s *thread;
bbl_interface_s *interface;
bbl_ethernet_header_s *eth;
bbl_stream_group_s *stream_group;
uint8_t *buf;
uint16_t buf_len;
code/bngblaster/src/io/io_interface.c
+8 -4
View File
@@ -15,7 +15,7 @@ set_kernel_info(bbl_interface_s *interface)
int fd = socket(PF_INET, SOCK_DGRAM, IPPROTO_IP);
snprintf(ifr.ifr_name, sizeof(ifr.ifr_name), "%s", interface->name);
if (ioctl(fd, SIOCGIFHWADDR, &ifr) == -1) {
if(ioctl(fd, SIOCGIFHWADDR, &ifr) == -1) {
LOG(ERROR, "Getting MAC address error %s (%d) for interface %s\n",
strerror(errno), errno, interface->name);
close(fd);
@@ -25,7 +25,7 @@ set_kernel_info(bbl_interface_s *interface)
memset(&ifr, 0, sizeof(ifr));
snprintf(ifr.ifr_name, sizeof(ifr.ifr_name), "%s", interface->name);
if (ioctl(fd, SIOCGIFINDEX, &ifr) == -1) {
if(ioctl(fd, SIOCGIFINDEX, &ifr) == -1) {
LOG(ERROR, "Get interface index error %s (%d) for interface %s\n",
strerror(errno), errno, interface->name);
close(fd);
@@ -175,8 +175,12 @@ io_interface_init(bbl_interface_s *interface)
bbl_link_config_s *config = interface->config;
if(config->io_mode != IO_MODE_DPDK) {
set_kernel_info(interface);
set_promisc(interface);
if(!set_kernel_info(interface)) {
return false;
}
if(!set_promisc(interface)) {
return false;
}
}
if(*(uint32_t*)config->mac) {
memcpy(interface->mac, config->mac, ETH_ADDR_LEN);
code/bngblaster/src/io/io_packet_mmap.c
+2 -2
View File
@@ -362,10 +362,10 @@ io_packet_mmap_init(io_handle_s *io)
}
} else {
if(io->direction == IO_INGRESS) {
timer_add_periodic(&g_ctx->timer_root, &interface->rx_job, "RX", 0,
timer_add_periodic(&g_ctx->timer_root, &interface->io.rx_job, "RX", 0,
config->rx_interval, io, &io_packet_mmap_rx_job);
} else {
timer_add_periodic(&g_ctx->timer_root, &interface->tx_job, "TX", 0,
timer_add_periodic(&g_ctx->timer_root, &interface->io.tx_job, "TX", 0,
config->tx_interval, io, &io_packet_mmap_tx_job);
}
}
code/bngblaster/src/io/io_raw.c
+2 -2
View File
@@ -199,10 +199,10 @@ io_raw_init(io_handle_s *io)
}
} else {
if(io->direction == IO_INGRESS) {
timer_add_periodic(&g_ctx->timer_root, &interface->rx_job, "RX", 0,
timer_add_periodic(&g_ctx->timer_root, &interface->io.rx_job, "RX", 0,
config->rx_interval, io, &io_raw_rx_job);
} else {
timer_add_periodic(&g_ctx->timer_root, &interface->tx_job, "TX", 0,
timer_add_periodic(&g_ctx->timer_root, &interface->io.tx_job, "TX", 0,
config->tx_interval, io, &io_raw_tx_job);
}
}
code/bngblaster/src/io/io_thread.c
+4 -4
View File
@@ -285,16 +285,16 @@ io_thread_init(io_handle_s *io)
thread->run_fn = io_thread_timer_loop;
/* Add thread main loop timers/jobs */
if(io->direction == IO_INGRESS && !interface->rx_job) {
if(io->direction == IO_INGRESS && !interface->io.rx_job) {
/** Start job reading from RX thread TXQ */
timer_add_periodic(&g_ctx->timer_root, &interface->rx_job, "RX",
timer_add_periodic(&g_ctx->timer_root, &interface->io.rx_job, "RX",
0, config->rx_interval,
interface, &io_thread_main_rx_job);
}
if(io->direction == IO_EGRESS && !interface->tx_job) {
if(io->direction == IO_EGRESS && !interface->io.tx_job) {
/** Start job writing to first TX thread TXQ */
timer_add_periodic(&g_ctx->timer_root, &interface->tx_job, "TX",
timer_add_periodic(&g_ctx->timer_root, &interface->io.tx_job, "TX",
0, config->tx_interval,
interface, &io_thread_main_tx_job);
}
code/common/src/logging.h
+1
View File
@@ -41,6 +41,7 @@ enum {
TCP,
LSDB,
LSP,
LAG,
LOG_ID_MAX
};