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

Flow Scaling (10M Flows)

Browse Source
This commit is contained in:
Christian Giese
2024-02-06 12:54:50 +00:00
committed by Hannes Gredler
parent 3c3cd25bc9
commit 0365835fbf
28 changed files with 982 additions and 947 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/bngblaster/src/bbl.c
+6 -3
View File
@@ -590,6 +590,12 @@ main(int argc, char *argv[])
}
LOG(INFO, "Total PPS of all streams: %.2f\n", g_ctx->total_pps);
if(!bbl_stream_index_init()) {
fprintf(stderr, "Error: Failed to init stream index\n");
goto CLEANUP;
}
/* Setup control job. */
timer_add_periodic(&g_ctx->timer_root, &g_ctx->control_timer, "Control Timer",
1, 0, g_ctx, &bbl_ctrl_job);
@@ -601,9 +607,6 @@ main(int argc, char *argv[])
}
}
/* Init IO stream token buckets. */
io_init_stream_token_bucket();
/* Start smear job. Use a crazy nsec bucket '12345678',
* such that we do not accidentally smear ourselves. */
timer_add_periodic(&g_ctx->timer_root, &g_ctx->smear_timer, "Timer Smearing",
code/bngblaster/src/bbl_a10nsp.c
+1 -1
View File
@@ -92,7 +92,7 @@ bbl_a10nsp_interfaces_add()
/* Timer to compute periodic rates */
timer_add_periodic(&g_ctx->timer_root, &a10nsp_interface->rate_job, "Rate Computation", 1, 0, a10nsp_interface,
&bbl_a10nsp_interface_rate_job);
a10nsp_interface->rate_job->reset = false;
LOG(DEBUG, "Added a10nsp interface %s\n", a10nsp_config->interface);
a10nsp_config = a10nsp_config->next;
}
code/bngblaster/src/bbl_access.c
+1 -1
View File
@@ -78,7 +78,7 @@ bbl_access_interfaces_add()
/* Timer to compute periodic rates */
timer_add_periodic(&g_ctx->timer_root, &access_interface->rate_job, "Rate Computation", 1, 0, access_interface,
&bbl_access_interface_rate_job);
access_interface->rate_job->reset = false;
LOG(DEBUG, "Added access interface %s\n", access_config->interface);
}
access_config->access_interface = interface->access;
code/bngblaster/src/bbl_config.c
+50 -30
View File
@@ -459,6 +459,7 @@ link_add(char *interface_name)
link_config = calloc(1, sizeof(bbl_link_config_s));
link_config->interface = strdup(interface_name);
link_config->io_mode = g_ctx->config.io_mode;
link_config->io_burst = g_ctx->config.io_burst;
link_config->io_slots_rx = g_ctx->config.io_slots;
link_config->io_slots_tx = g_ctx->config.io_slots;
link_config->qdisc_bypass = g_ctx->config.qdisc_bypass;
@@ -479,11 +480,12 @@ json_parse_link(json_t *link, bbl_link_config_s *link_config)
const char *schema[] = {
"interface", "description", "mac",
"io-mode", "io-slots", "io-slots-tx",
"io-slots-rx", "qdisc-bypass", "tx-interval",
"rx-interval", "tx-threads", "rx-threads",
"rx-cpuset", "tx-cpuset", "lag-interface",
"lacp-priority"
"io-mode", "io-slots" "io-burst",
"io-slots-tx", "io-slots-rx", "qdisc-bypass",
"tx-interval","rx-interval",
"tx-threads", "rx-threads",
"rx-cpuset", "tx-cpuset",
"lag-interface", "lacp-priority"
};
if(!schema_validate(link, "links", schema,
sizeof(schema)/sizeof(schema[0]))) {
@@ -553,7 +555,12 @@ json_parse_link(json_t *link, bbl_link_config_s *link_config)
if(value) {
link_config->io_slots_rx = json_number_value(value);
}
JSON_OBJ_GET_NUMBER(link, value, "links", "io-burst", 1, 65535);
if(value) {
link_config->io_burst = json_number_value(value);
} else {
link_config->io_burst = g_ctx->config.io_burst;
}
JSON_OBJ_GET_BOOL(link, value, "links", "qdisc-bypass");
if(value) {
link_config->qdisc_bypass = json_boolean_value(value);
@@ -2538,31 +2545,38 @@ json_parse_config_streams(json_t *root)
bbl_stream_config_s *stream_config = g_ctx->config.stream_config;
if(json_typeof(root) != JSON_OBJECT) {
fprintf(stderr, "JSON config error: Configuration root element must object\n");
fprintf(stderr, "JSON config error: Configuration root element must be of type object\n");
return false;
}
section = json_object_get(root, "streams");
if(json_is_array(section)) {
/* Get tail end of stream-config list. */
if(stream_config) {
while(stream_config->next) {
stream_config = stream_config->next;
}
if(section) {
if(!json_is_array(section)) {
fprintf(stderr, "JSON config error: Configuration streams element must contain list of objects\n");
return false;
}
/* Config is provided as array (multiple streams) */
size = json_array_size(section);
for(i = 0; i < size; i++) {
if(!stream_config) {
g_ctx->config.stream_config = calloc(1, sizeof(bbl_stream_config_s));
stream_config = g_ctx->config.stream_config;
} else {
stream_config->next = calloc(1, sizeof(bbl_stream_config_s));
stream_config = stream_config->next;
}
if(!json_parse_stream(json_array_get(section, i), stream_config)) {
return false;
}
} else {
return true;
}
/* Get tail end of stream-config list. */
if(stream_config) {
while(stream_config->next) {
stream_config = stream_config->next;
}
}
/* Config is provided as array (multiple streams) */
size = json_array_size(section);
for(i = 0; i < size; i++) {
if(!stream_config) {
g_ctx->config.stream_config = calloc(1, sizeof(bbl_stream_config_s));
stream_config = g_ctx->config.stream_config;
} else {
stream_config->next = calloc(1, sizeof(bbl_stream_config_s));
stream_config = stream_config->next;
}
if(!json_parse_stream(json_array_get(section, i), stream_config)) {
return false;
}
}
return true;
@@ -3571,7 +3585,7 @@ json_parse_config(json_t *root)
if(json_is_object(section)) {
const char *schema[] = {
"io-mode", "io-slots", "qdisc-bypass",
"io-mode", "io-slots", "io-burst", "qdisc-bypass",
"tx-interval", "rx-interval", "tx-threads",
"rx-threads", "capture-include-streams", "mac-modifier",
"lag", "network", "access", "a10nsp", "links"
@@ -3608,6 +3622,11 @@ json_parse_config(json_t *root)
if(value) {
g_ctx->config.io_slots = json_number_value(value);
}
value = json_object_get(section, "io-burst");
JSON_OBJ_GET_NUMBER(section, value, "interfaces", "io-burst", 1, 65535);
if(value) {
g_ctx->config.io_burst = json_number_value(value);
}
JSON_OBJ_GET_BOOL(section, value, "interfaces", "qdisc-bypass");
if(value) {
g_ctx->config.qdisc_bypass = json_boolean_value(value);
@@ -4041,9 +4060,10 @@ bbl_config_init_defaults()
g_ctx->pcap.include_streams = false;
g_ctx->config.username = g_default_user;
g_ctx->config.password = g_default_pass;
g_ctx->config.tx_interval = 1 * MSEC;
g_ctx->config.rx_interval = 1 * MSEC;
g_ctx->config.tx_interval = 0.1 * MSEC;
g_ctx->config.rx_interval = 0.1 * MSEC;
g_ctx->config.io_slots = 4096;
g_ctx->config.io_burst = 256;
g_ctx->config.io_max_stream_len = 9000;
g_ctx->config.qdisc_bypass = true;
g_ctx->config.sessions = 1;
@@ -4099,7 +4119,7 @@ bbl_config_init_defaults()
g_ctx->config.multicast_traffic_pps = 1000;
g_ctx->config.traffic_autostart = true;
g_ctx->config.stream_rate_calc = true;
g_ctx->config.stream_max_burst = 32;
g_ctx->config.stream_max_burst = 16;
g_ctx->config.multicast_traffic_autostart = true;
g_ctx->config.session_traffic_autostart = true;
}
code/bngblaster/src/bbl_config.h
+1
View File
@@ -149,6 +149,7 @@ typedef struct bbl_link_config_
uint16_t io_slots_tx;
uint16_t io_slots_rx;
uint16_t io_burst;
bool qdisc_bypass;
code/bngblaster/src/bbl_ctx.c
+4 -4
View File
@@ -80,7 +80,6 @@ bbl_ctx_add()
g_ctx->vlan_session_dict = hashtable_dict_new((dict_compare_func)bbl_compare_key64, bbl_key64_hash, BBL_SESSION_HASHTABLE_SIZE);
g_ctx->l2tp_session_dict = hashtable_dict_new((dict_compare_func)bbl_compare_key32, bbl_key32_hash, BBL_SESSION_HASHTABLE_SIZE);
g_ctx->li_flow_dict = hashtable_dict_new((dict_compare_func)bbl_compare_key32, bbl_key32_hash, BBL_LI_HASHTABLE_SIZE);
g_ctx->stream_flow_dict = hashtable_dict_new((dict_compare_func)bbl_compare_key64, bbl_key64_hash, BBL_STREAM_FLOW_HASHTABLE_SIZE);
return true;
}
@@ -119,13 +118,14 @@ bbl_ctx_del() {
bbl_session_free(p);
}
}
free(g_ctx->session_list);
if(g_ctx->session_list) free(g_ctx->session_list);
if(g_ctx->stream_index) free(g_ctx->stream_index);
/* Free hash table dictionaries. */
dict_free(g_ctx->vlan_session_dict, NULL);
dict_free(g_ctx->l2tp_session_dict, NULL);
dict_free(g_ctx->li_flow_dict, NULL);
dict_free(g_ctx->stream_flow_dict, NULL);
pcapng_free();
free(g_ctx);
code/bngblaster/src/bbl_ctx.h
+7 -1
View File
@@ -78,7 +78,11 @@ 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 */
bbl_stream_s **stream_index;
bbl_stream_s *stream_head;
bbl_stream_s *stream_tail;
uint64_t streams;
bbl_stream_group_s *stream_groups;
@@ -89,6 +93,7 @@ typedef struct bbl_ctx_
char *ctrl_socket_path;
bbl_ctrl_thread_s *ctrl_thread;
io_thread_s *io_threads; /* single linked list of threads */
io_bucket_s *io_bucket;
bool tcp;
bool dpdk;
@@ -146,6 +151,7 @@ typedef struct bbl_ctx_
io_mode_t io_mode;
uint16_t io_slots;
uint16_t io_burst;
uint16_t io_max_stream_len;
bool qdisc_bypass;
code/bngblaster/src/bbl_def.h
-1
View File
@@ -53,7 +53,6 @@
#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 128993 /* is a prime number */
/* Mock Addresses */
#define MOCK_IP_LOCAL 167772170 /* 10.0.0.10 */
code/bngblaster/src/bbl_lag.h
+1 -1
View File
@@ -20,7 +20,7 @@ typedef struct bbl_lag_
uint8_t active_count;
bbl_lag_member_s *active_list[LAG_MEMBER_ACTIVE_MAX];
uint64_t select;
uint32_t select;
CIRCLEQ_ENTRY(bbl_lag_) lag_qnode;
CIRCLEQ_HEAD(lag_member_, bbl_lag_member_ ) lag_member_qhead; /* list of member interfaces */
code/bngblaster/src/bbl_network.c
+1 -1
View File
@@ -207,7 +207,7 @@ bbl_network_interfaces_add()
/* Timer to compute periodic rates */
timer_add_periodic(&g_ctx->timer_root, &network_interface->rate_job, "Rate Computation", 1, 0, network_interface,
&bbl_network_interface_rate_job);
network_interface->rate_job->reset = false;
LOG(DEBUG, "Added network interface %s\n", ifname);
network_config = network_config->next;
}
code/bngblaster/src/bbl_protocols.c
+2 -2
View File
@@ -3536,8 +3536,8 @@ decode_ipv4(uint8_t *buf, uint16_t len,
return DECODE_ERROR;
}
ipv4->id = be16toh(header->ip_id);
ipv4->offset = be16toh(header->ip_off);
if(header->ip_id) ipv4->id = be16toh(header->ip_id);
if(header->ip_off) ipv4->offset = be16toh(header->ip_off);
ipv4->ttl = header->ip_ttl;
ipv4->protocol = header->ip_p;
code/bngblaster/src/bbl_protocols.h
+1
View File
@@ -309,6 +309,7 @@ typedef enum protocol_error_ {
IGNORED,
EMPTY,
FULL,
STREAM_WAIT
} protocol_error_t;
typedef enum icmpv6_message_ {
code/bngblaster/src/bbl_rx.c
+16 -41
View File
@@ -14,10 +14,8 @@ bbl_rx_stream_network(bbl_network_interface_s *interface,
bbl_ethernet_header_s *eth)
{
bbl_stream_s *stream;
if(!eth->bbl || memcmp(interface->mac, eth->dst, ETH_ADDR_LEN) != 0) {
return false;
}
stream = bbl_stream_rx(eth, NULL);
if(!eth->bbl) return false;
stream = bbl_stream_rx(eth, interface->mac);
if(stream) {
if(stream->rx_network_interface == NULL) {
stream->rx_network_interface = interface;
@@ -32,29 +30,13 @@ bbl_rx_stream_access(bbl_access_interface_s *interface,
bbl_ethernet_header_s *eth)
{
bbl_stream_s *stream;
bbl_session_s *session;
uint32_t session_id = 0;
if(!(eth->bbl && eth->bbl->type == BBL_TYPE_UNICAST)) {
return false;
}
session_id |= eth->dst[5];
session_id |= eth->dst[4] << 8;
session_id |= eth->dst[3] << 16;
session = bbl_session_get(session_id);
if(session) {
if(session->session_state != BBL_TERMINATED &&
session->session_state != BBL_IDLE) {
stream = bbl_stream_rx(eth, session);
if(stream) {
if(stream->rx_access_interface == NULL) {
stream->rx_access_interface = interface;
}
return true;
}
if(!eth->bbl) return false;
stream = bbl_stream_rx(eth, NULL);
if(stream) {
if(stream->rx_access_interface == NULL) {
stream->rx_access_interface = interface;
}
return true;
}
return false;
}
@@ -64,10 +46,8 @@ bbl_rx_stream_a10nsp(bbl_a10nsp_interface_s *interface,
bbl_ethernet_header_s *eth)
{
bbl_stream_s *stream;
if(!eth->bbl || memcmp(interface->mac, eth->dst, ETH_ADDR_LEN) != 0) {
return false;
}
stream = bbl_stream_rx(eth, NULL);
if(!eth->bbl) return false;
stream = bbl_stream_rx(eth, interface->mac);
if(stream) {
if(stream->rx_a10nsp_interface == NULL) {
stream->rx_a10nsp_interface = interface;
@@ -81,19 +61,14 @@ bool
bbl_rx_thread(bbl_interface_s *interface,
bbl_ethernet_header_s *eth)
{
bbl_network_interface_s *network_interface = interface->network;
bbl_network_interface_s *network_interface;
if(interface->state == INTERFACE_DISABLED) {
return true;
}
while(network_interface) {
if(network_interface->vlan == eth->vlan_outer) {
return bbl_rx_stream_network(network_interface, eth);
}
network_interface = network_interface->next;
}
if(interface->access) {
network_interface = interface->network_vlan[eth->vlan_outer];
if(network_interface) {
return bbl_rx_stream_network(network_interface, eth);
} else if(interface->access) {
return bbl_rx_stream_access(interface->access, eth);
} else if(interface->a10nsp) {
return bbl_rx_stream_a10nsp(interface->a10nsp, eth);
@@ -130,7 +105,7 @@ bbl_rx_handler(bbl_interface_s *interface,
if(!bbl_rx_stream_network(network_interface, eth)) {
bbl_network_rx_handler(network_interface, eth);
}
} else if(interface->access) {
} else if(interface->access) {
if(!bbl_rx_stream_access(interface->access, eth)) {
bbl_access_rx_handler(interface->access, eth);
}
code/bngblaster/src/bbl_session.c
+3 -14
View File
@@ -280,10 +280,6 @@ bbl_session_monkey_job(timer_s *timer) {
/**
* bbl_session_get
*
* This function allows to change the state of a session including
* the required action caused by state changes.
*
* @param ctx global context
* @param session_id session-id
* @return session or NULL if session not found
*/
@@ -1749,25 +1745,18 @@ bbl_session_ctrl_traffic_stop(int fd, uint32_t session_id, json_t *arguments __a
int
bbl_session_ctrl_traffic_reset(int fd, uint32_t session_id __attribute__((unused)), json_t *arguments __attribute__((unused)))
{
bbl_stream_s *stream;
struct dict_itor *itor;
bbl_stream_s *stream = g_ctx->stream_head;
g_ctx->stats.session_traffic_flows_verified = 0;
/* Iterate over all traffic streams */
itor = dict_itor_new(g_ctx->stream_flow_dict);
dict_itor_first(itor);
for (; dict_itor_valid(itor); dict_itor_next(itor)) {
stream = (bbl_stream_s*)*dict_itor_datum(itor);
if(!stream) {
continue;
}
while(stream) {
if(stream->session && stream->session_traffic) {
stream->session->session_traffic.flows_verified = 0;
bbl_stream_reset(stream);
}
stream = stream->next;
}
dict_itor_free(itor);
return bbl_ctrl_status(fd, "ok", 200, NULL);
}
code/bngblaster/src/bbl_stats.c
+32 -53
View File
@@ -165,8 +165,6 @@ bbl_stats_generate(bbl_stats_s * stats)
bbl_stats_update_cps();
bbl_stats_generate_multicast(stats, false);
struct dict_itor *itor;
/* Iterate over all sessions */
for(i = 0; i < g_ctx->sessions; i++) {
session = &g_ctx->session_list[i];
@@ -370,36 +368,32 @@ bbl_stats_generate(bbl_stats_s * stats)
}
/* Iterate over all traffic streams */
itor = dict_itor_new(g_ctx->stream_flow_dict);
dict_itor_first(itor);
for (; dict_itor_valid(itor); dict_itor_next(itor)) {
stream = (bbl_stream_s*)*dict_itor_datum(itor);
if(stream) {
if(stats->min_stream_loss) {
if(stream->rx_loss < stats->min_stream_loss) stats->min_stream_loss = stream->rx_loss;
} else {
stats->min_stream_loss = stream->rx_loss;
}
if(stream->rx_loss > stats->max_stream_loss) stats->max_stream_loss = stream->rx_loss;
if(stream->rx_first_seq) {
if(stats->min_stream_rx_first_seq) {
if(stream->rx_first_seq < stats->min_stream_rx_first_seq) stats->min_stream_rx_first_seq = stream->rx_first_seq;
} else {
stats->min_stream_rx_first_seq = stream->rx_first_seq;
}
if(stream->rx_first_seq > stats->max_stream_rx_first_seq) stats->max_stream_rx_first_seq = stream->rx_first_seq;
if(stats->min_stream_delay_us) {
if(stream->rx_min_delay_us < stats->min_stream_delay_us) stats->min_stream_delay_us = stream->rx_min_delay_us;
} else {
stats->min_stream_delay_us = stream->rx_min_delay_us;
}
if(stream->rx_max_delay_us > stats->max_stream_delay_us) stats->max_stream_delay_us = stream->rx_max_delay_us;
}
stream = g_ctx->stream_head;
while(stream) {
if(stats->min_stream_loss) {
if(stream->rx_loss < stats->min_stream_loss) stats->min_stream_loss = stream->rx_loss;
} else {
stats->min_stream_loss = stream->rx_loss;
}
if(stream->rx_loss > stats->max_stream_loss) stats->max_stream_loss = stream->rx_loss;
if(stream->rx_first_seq) {
if(stats->min_stream_rx_first_seq) {
if(stream->rx_first_seq < stats->min_stream_rx_first_seq) stats->min_stream_rx_first_seq = stream->rx_first_seq;
} else {
stats->min_stream_rx_first_seq = stream->rx_first_seq;
}
if(stream->rx_first_seq > stats->max_stream_rx_first_seq) stats->max_stream_rx_first_seq = stream->rx_first_seq;
if(stats->min_stream_delay_us) {
if(stream->rx_min_delay_us < stats->min_stream_delay_us) stats->min_stream_delay_us = stream->rx_min_delay_us;
} else {
stats->min_stream_delay_us = stream->rx_min_delay_us;
}
if(stream->rx_max_delay_us > stats->max_stream_delay_us) stats->max_stream_delay_us = stream->rx_max_delay_us;
}
stream = stream->next;
}
dict_itor_free(itor);
}
void
@@ -805,8 +799,6 @@ bbl_stats_json(bbl_stats_s * stats)
bbl_session_s *session;
bbl_stream_s *stream;
struct dict_itor *itor;
json_t *root = NULL;
json_t *jobj = NULL;
json_t *jobj_array = NULL;
@@ -1204,18 +1196,14 @@ bbl_stats_json(bbl_stats_s * stats)
if(g_ctx->config.json_report_streams) {
jobj_array = json_array();
itor = dict_itor_new(g_ctx->stream_flow_dict);
dict_itor_first(itor);
for (; dict_itor_valid(itor); dict_itor_next(itor)) {
stream = (bbl_stream_s*)*dict_itor_datum(itor);
if(stream) {
jobj_sub = bbl_stream_json(stream);
if(jobj_sub) {
json_array_append(jobj_array, jobj_sub);
}
stream = g_ctx->stream_head;
while(stream) {
jobj_sub = bbl_stream_json(stream);
if(jobj_sub) {
json_array_append(jobj_array, jobj_sub);
}
stream = stream->next;
}
dict_itor_free(itor);
json_object_set(jobj, "streams", jobj_array);
}
@@ -1233,7 +1221,6 @@ void
bbl_compute_avg_rate(bbl_rate_s *rate, uint64_t current_value)
{
uint8_t idx;
uint64_t div;
uint64_t sum;
if(current_value == 0) return;
@@ -1241,18 +1228,10 @@ bbl_compute_avg_rate(bbl_rate_s *rate, uint64_t current_value)
rate->diff_value[rate->cursor] = current_value - rate->last_value;
sum = 0;
div = 0;
for(idx = 0; idx < BBL_AVG_SAMPLES; idx++) {
if(rate->diff_value[idx]) {
sum += rate->diff_value[idx];
div++;
}
}
if(div) {
rate->avg = sum / div;
} else {
rate->avg = 0;
sum += rate->diff_value[idx];
}
rate->avg = sum / BBL_AVG_SAMPLES;
if(rate->avg > rate->avg_max) {
rate->avg_max = rate->avg;
}
code/bngblaster/src/bbl_stream.c
+336 -361
View File
File diff suppressed because it is too large Load Diff
code/bngblaster/src/bbl_stream.h
+53 -44
View File
@@ -67,6 +67,7 @@ typedef struct bbl_stream_config_
typedef struct bbl_stream_group_
{
double pps;
uint32_t count;
bbl_stream_s *head;
struct timer_ *timer;
@@ -82,38 +83,9 @@ typedef struct bbl_stream_
uint8_t sub_type;
uint8_t direction;
bbl_stream_config_s *config;
bbl_stream_group_s *group;
bbl_stream_s *group_next; /* Next stream of same group */
bbl_stream_s *reverse; /* Reverse stream direction */
uint32_t session_version;
bbl_session_s *session;
bbl_stream_s *session_next; /* Next stream of same session */
endpoint_state_t *endpoint;
io_handle_s *io;
bbl_access_interface_s *access_interface;
bbl_network_interface_s *network_interface;
bbl_a10nsp_interface_s *a10nsp_interface;
struct timer_ *tx_timer;
bbl_interface_s *tx_interface; /* TX interface */
uint8_t *tx_buf; /* TX buffer */
uint16_t tx_len; /* TX length */
uint16_t tx_bbl_hdr_len; /* TX BBL HDR length */
uint64_t tx_interval; /* TX interval in nsec */
__time_t tx_first_epoch;
uint32_t ipv4_src;
uint32_t ipv4_dst;
uint8_t *ipv6_src;
uint8_t *ipv6_dst;
bool threaded;
bool session_traffic;
bool active;
bool setup;
bool verified;
bool wait;
@@ -123,22 +95,50 @@ typedef struct bbl_stream_
bool tcp;
bool lag;
bool ldp_lookup;
uint32_t session_version;
uint32_t ldp_entry_version;
uint32_t ipv4_src;
uint32_t ipv4_dst;
uint16_t tx_len; /* TX length */
uint16_t tx_bbl_hdr_len; /* TX BBL HDR length */
uint8_t *tx_buf; /* TX buffer */
uint8_t *ipv6_src;
uint8_t *ipv6_dst;
bbl_stream_config_s *config;
bbl_stream_s *next; /* Next stream (global) */
bbl_stream_s *io_next; /* Next stream of same IO handle */
bbl_stream_s *group_next; /* Next stream of same group */
bbl_stream_s *session_next; /* Next stream of same session */
bbl_stream_s *reverse; /* Reverse stream direction */
bbl_stream_group_s *group;
bbl_session_s *session;
endpoint_state_t *endpoint;
io_handle_s *io;
io_bucket_s *io_bucket;
bbl_access_interface_s *access_interface;
bbl_network_interface_s *network_interface;
bbl_a10nsp_interface_s *a10nsp_interface;
bbl_interface_s *tx_interface; /* TX interface */
ldp_db_entry_s *ldp_entry;
bool send_window_active;
uint64_t send_window_start_packets;
struct timespec send_window_start;
struct timespec wait_start;
CIRCLEQ_ENTRY(bbl_stream_) tx_qnode;
uint32_t token_bucket;
uint32_t token_burst;
uint64_t lag_select;
uint64_t tx_packets;
uint64_t tokens;
uint64_t tokens_burst;
uint32_t lag_select;
__time_t tx_first_epoch;
struct timespec wait_start;
char _pad0 __attribute__((__aligned__(CACHE_LINE_SIZE))); /* empty cache line */
@@ -194,6 +194,12 @@ typedef struct bbl_stream_
} bbl_stream_s;
bbl_stream_s *
bbl_stream_index_get(uint64_t flow_id);
bool
bbl_stream_index_init();
bool
bbl_stream_session_init(bbl_session_s *session);
@@ -204,10 +210,13 @@ void
bbl_stream_final();
protocol_error_t
bbl_stream_tx(io_handle_s *io, uint8_t *buf, uint16_t *len);
bbl_stream_io_send(io_handle_s *io, bbl_stream_s *stream);
bbl_stream_s*
bbl_stream_io_send_iter(io_handle_s *io);
bbl_stream_s *
bbl_stream_rx(bbl_ethernet_header_s *eth, bbl_session_s *session);
bbl_stream_rx(bbl_ethernet_header_s *eth, uint8_t *mac);
void
bbl_stream_reset(bbl_stream_s *stream);
code/bngblaster/src/io/io.c
+1 -33
View File
@@ -7,36 +7,4 @@
* SPDX-License-Identifier: BSD-3-Clause
*/
#include "io.h"
void
io_update_stream_token_bucket(io_handle_s *io)
{
io->stream_tokens += io->stream_rate;
if(io->stream_tokens > io->stream_burst) {
io->stream_tokens = io->stream_burst;
}
}
void
io_init_stream_token_bucket()
{
bbl_interface_s *interface;
io_handle_s *io;
double rate;
CIRCLEQ_FOREACH(interface, &g_ctx->interface_qhead, interface_qnode) {
io = interface->io.tx;
while(io) {
rate = (io->stream_pps / (io->interface->config->tx_interval / 1000)) * 1.3;
io->stream_rate = rate;
if(rate - (double)io->stream_rate) {
/* Roundup. */
io->stream_rate++;
}
io->stream_tokens = 0;
io->stream_burst = io->stream_rate * 3;
io = io->next;
}
}
}
#include "io.h"
code/bngblaster/src/io/io.h
+1 -6
View File
@@ -19,6 +19,7 @@
#include "../bbl_txq.h"
#include "io_def.h"
#include "io_bucket.h"
#include "io_socket.h"
#include "io_interface.h"
#include "io_thread.h"
@@ -30,10 +31,4 @@
#include "io_dpdk.h"
#endif
void
io_update_stream_token_bucket(io_handle_s *io);
void
io_init_stream_token_bucket();
#endif
code/bngblaster/src/io/io_bucket.c
+56
View File
@@ -0,0 +1,56 @@
/*
* BNG Blaster (BBL) - IO Bucket
*
* Christian Giese, January 2024
*
* Copyright (C) 2020-2023, RtBrick, Inc.
* SPDX-License-Identifier: BSD-3-Clause
*/
#include "io.h"
void
io_tocken_job(timer_s *timer)
{
io_bucket_s *io_bucket = timer->data;
struct timespec time_elapsed;
uint64_t tokens;
if(unlikely(!io_bucket->timestamp_start.tv_sec)) {
io_bucket->timestamp_start.tv_sec = timer->timestamp->tv_sec;
io_bucket->timestamp_start.tv_nsec = timer->timestamp->tv_nsec;
return;
}
timespec_sub(&time_elapsed, timer->timestamp, &io_bucket->timestamp_start);
tokens = io_bucket->tokens_per_sec * time_elapsed.tv_sec;
tokens += (io_bucket->tokens_per_sec * time_elapsed.tv_nsec) / SEC;
io_bucket->tokens = tokens;
}
static io_bucket_s *
io_bucket_add(double pps)
{
io_bucket_s *io_bucket = calloc(1, sizeof(io_bucket_s));
io_bucket->next = g_ctx->io_bucket;
g_ctx->io_bucket = io_bucket;
io_bucket->pps = pps;
io_bucket->tokens_per_sec = pps * IO_TOKENS_PER_PACKET;
timer_add_periodic(&g_ctx->timer_root, &io_bucket->timer,
"TB", 0, 1*MSEC+1, io_bucket, &io_tocken_job);
return io_bucket;
}
void
io_bucket_stream(bbl_stream_s *stream)
{
io_bucket_s *io_bucket = g_ctx->io_bucket;
while(io_bucket) {
if(io_bucket->pps == stream->config->pps) {
break;
}
io_bucket = io_bucket->next;
}
if(!io_bucket) {
io_bucket = io_bucket_add(stream->config->pps);
}
stream->io_bucket = io_bucket;
}
code/bngblaster/src/io/io_bucket.h
+15
View File
@@ -0,0 +1,15 @@
/*
* BNG Blaster (BBL) - IO Bucket
*
* Christian Giese, January 2024
*
* Copyright (C) 2020-2023, RtBrick, Inc.
* SPDX-License-Identifier: BSD-3-Clause
*/
#ifndef __BBL_IO_BUCKET_H__
#define __BBL_IO_BUCKET_H__
void
io_bucket_stream(bbl_stream_s *stream);
#endif
code/bngblaster/src/io/io_def.h
+14 -12
View File
@@ -9,6 +9,8 @@
#ifndef __BBL_IO_DEF_H__
#define __BBL_IO_DEF_H__
#define IO_TOKENS_PER_PACKET 1000
typedef struct io_handle_ io_handle_s;
typedef struct io_thread_ io_thread_s;
@@ -38,6 +40,15 @@ typedef enum {
IO_MODE_AF_XDP /* AF_XDP */
} __attribute__ ((__packed__)) io_mode_t;
typedef struct io_bucket_ {
double pps;
uint64_t tokens_per_sec;
uint64_t tokens;
struct timer_ *timer;
struct timespec timestamp_start;
struct io_bucket_ *next;
} io_bucket_s;
typedef struct io_handle_ {
io_mode_t mode;
io_direction_t direction;
@@ -70,10 +81,9 @@ typedef struct io_handle_ {
uint16_t vlan_tpid;
double stream_pps;
uint32_t stream_tokens;
uint32_t stream_rate;
uint32_t stream_burst;
CIRCLEQ_HEAD(stream_tx_, bbl_stream_) stream_tx_qhead;
uint32_t stream_count;
bbl_stream_s *stream_head;
bbl_stream_s *stream_cur;
struct timespec timestamp; /* user space timestamps */
@@ -93,7 +103,6 @@ typedef struct io_handle_ {
} io_handle_s;
typedef void (*io_thread_cb_fn)(io_thread_s *thread);
typedef bool (*io_thread_stream_cb_fn)(bbl_stream_s *stream);
typedef struct io_thread_ {
pthread_t thread;
@@ -108,18 +117,11 @@ typedef struct io_thread_ {
io_thread_cb_fn run_fn;
io_thread_cb_fn teardown_fn;
io_thread_stream_cb_fn stream_tx_fn;
uint8_t *sp;
io_handle_s *io;
bbl_txq_s *txq;
struct {
struct timer_root_ root;
struct timer_ *io;
} timer;
struct io_thread_ *next;
} io_thread_s;
code/bngblaster/src/io/io_dpdk.c
+158 -128
View File
@@ -16,9 +16,18 @@
#ifdef BNGBLASTER_DPDK
#include <dev_driver.h>
#include <rte_memory.h>
#include <rte_launch.h>
#include <rte_eal.h>
#include <rte_common.h>
#include <rte_dev.h>
#include <rte_bbdev.h>
#include <rte_cycles.h>
#include <rte_random.h>
#include <rte_hexdump.h>
#include <rte_interrupts.h>
#include <rte_per_lcore.h>
#include <rte_lcore.h>
#include <rte_debug.h>
@@ -26,14 +35,14 @@
#include <rte_ethdev.h>
#include <rte_malloc.h>
#define NUM_MBUFS 4096
#define MBUF_CACHE_SIZE 250
#define BURST_SIZE_RX 64
#define NUM_MBUFS 8192
#define MBUF_CACHE_SIZE 256
#define BURST_SIZE_RX 256
#define BURST_SIZE_TX 32
static struct rte_eth_conf port_conf = {
.rxmode = {
.split_hdr_size = 0,
.mq_mode = 0,
},
.txmode = {
.mq_mode = RTE_ETH_MQ_TX_NONE,
@@ -66,7 +75,7 @@ io_dpdk_link_status(uint16_t port_id)
interface->state = INTERFACE_UP;
LOG(DPDK, "DPDK: interface %s (%u) link up (speed %u Mbps %s)\n",
interface->name, port_id, (unsigned)link.link_speed,
(link.link_duplex == ETH_LINK_FULL_DUPLEX) ? ("full-duplex") : ("half-duplex"));
(link.link_duplex == RTE_ETH_LINK_FULL_DUPLEX) ? ("full-duplex") : ("half-duplex"));
} else {
interface->state = INTERFACE_DOWN;
LOG(DPDK, "DPDK: interface %s (%u) link down\n",
@@ -194,17 +203,25 @@ io_dpdk_rx_job(timer_s *timer)
/* Copy RX timestamp */
eth->timestamp.tv_sec = io->timestamp.tv_sec;
eth->timestamp.tv_nsec = io->timestamp.tv_nsec;
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
bbl_rx_handler(interface, eth);
} else if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
}
rte_pktmbuf_free(packet);
}
@@ -245,73 +262,87 @@ io_dpdk_tx_job(timer_s *timer)
io_handle_s *io = timer->data;
bbl_interface_s *interface = io->interface;
uint32_t stream_packets = 0;
bool ctrl = true;
bbl_stream_s *stream = NULL;
uint16_t burst = interface->config->io_burst;
bool pcap = false;
assert(io->mode == IO_MODE_DPDK);
assert(io->direction == IO_EGRESS);
assert(io->thread == NULL);
io_update_stream_token_bucket(io);
/* Get TX timestamp */
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
while(true) {
/* If sendto fails, the failed packet remains in TX buffer to be retried
* in the next interval. */
if(io->buf_len) {
if(packet_is_bbl(io->buf, io->buf_len)) {
/* Update timestamp if BBL traffic is retried. */
*(uint32_t*)(io->buf + (io->buf_len - 8)) = io->timestamp.tv_sec;
*(uint32_t*)(io->buf + (io->buf_len - 4)) = io->timestamp.tv_nsec;
}
} else {
if(!io->mbuf) {
if(!io_dpdk_mbuf_alloc(io)) {
break;
}
}
if(ctrl) {
/* First send all control traffic which has higher priority. */
if(bbl_tx(interface, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
io->buf_len = 0;
ctrl = false;
continue;
}
} else {
/* Send traffic streams up to allowed burst. */
if(++stream_packets > io->stream_burst) {
break;
}
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
break;
}
while(burst) {
if(!io->mbuf) {
if(!io_dpdk_mbuf_alloc(io)) {
break;
}
}
io->mbuf->data_len = io->buf_len;
if(likely(io->buf_len == 0)) {
if(bbl_tx(interface, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
io->buf_len = 0;
break;
}
}
/* Transmit the packet. */
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) == 0) {
io->mbuf->data_len = io->buf_len;
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) != 0) {
/* Dump the packet into pcap file. */
if(unlikely(g_ctx->pcap.write_buf != NULL)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_OUTBOUND);
}
io->stats.packets++;
io->stats.bytes += io->buf_len;
io->mbuf = NULL;
io->buf_len = 0;
burst--;
} else {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
if(pcap) {
pcapng_fflush();
io->stats.io_errors++;
burst = 0;
}
}
while(burst) {
/* Send traffic streams up to allowed burst. */
if(!io->mbuf) {
if(!io_dpdk_mbuf_alloc(io)) {
break;
}
return;
}
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf && (ctrl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_OUTBOUND);
stream = bbl_stream_io_send_iter(io);
if(unlikely(stream == NULL)) {
break;
}
/* Transmit the packet. */
io->mbuf->data_len = io->buf_len;
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) != 0) {
/* Dump the packet into pcap file. */
if(unlikely(g_ctx->pcap.write_buf && g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_OUTBOUND);
}
stream->tx_packets++;
stream->flow_seq++;
io->stats.packets++;
io->stats.bytes += io->buf_len;
io->mbuf = NULL;
io->buf_len = 0;
burst--;
} else {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
io->stats.io_errors++;
burst = 0;
}
io->stats.packets++;
io->stats.bytes += io->buf_len;
io->mbuf = NULL;
io->buf = 0;
io->buf_len = 0;
}
if(pcap) {
pcapng_fflush();
@@ -337,12 +368,11 @@ io_dpdk_thread_rx_run_fn(io_thread_s *thread)
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 0;
sleep.tv_nsec = 10;
while(thread->active) {
nb_rx = rte_eth_rx_burst(port_id, io->queue, pkts_burst, BURST_SIZE_RX);
if(nb_rx == 0) {
sleep.tv_nsec = 10;
nanosleep(&sleep, &rem);
continue;
}
@@ -360,84 +390,89 @@ io_dpdk_thread_rx_run_fn(io_thread_s *thread)
}
}
/**
* This job is for DPDK TX in worker thread!
*/
void
io_dpdk_thread_tx_job(timer_s *timer)
io_dpdk_thread_tx_run_fn(io_thread_s *thread)
{
io_thread_s *thread = timer->data;
io_handle_s *io = thread->io;
bbl_interface_s *interface = io->interface;
bbl_txq_s *txq = thread->txq;
bbl_txq_slot_t *slot;
uint32_t stream_packets = 0;
bool ctrl = true;
bbl_stream_s *stream = NULL;
uint16_t io_burst = interface->config->io_burst;
uint16_t burst = 0;
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 1000;
assert(io->mode == IO_MODE_DPDK);
assert(io->direction == IO_EGRESS);
assert(io->thread);
io_update_stream_token_bucket(io);
while(thread->active) {
nanosleep(&sleep, &rem);
burst = io_burst;
/* Get TX timestamp */
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
while(true) {
/* If sendto fails, the failed packet remains in TX buffer to be retried
* in the next interval. */
if(io->buf_len) {
if(packet_is_bbl(io->buf, io->buf_len)) {
/* Update timestamp if BBL traffic is retried. */
*(uint32_t*)(io->buf + (io->buf_len - 8)) = io->timestamp.tv_sec;
*(uint32_t*)(io->buf + (io->buf_len - 4)) = io->timestamp.tv_nsec;
}
} else {
/* First send all control traffic which has higher priority. */
while((slot = bbl_txq_read_slot(txq))) {
/* This packet will be retried next interval
* because slot is not marked as read. */
if(!io->mbuf) {
if(!io_dpdk_mbuf_alloc(io)) {
break;
}
}
if(ctrl) {
/* First send all control traffic which has higher priority. */
slot = bbl_txq_read_slot(txq);
if(slot) {
io->buf_len = slot->packet_len;
memcpy(io->buf, slot->packet, slot->packet_len);
bbl_txq_read_next(txq);
} else {
ctrl = false;
continue;
}
/* Transmit the packet. */
io->mbuf->data_len = slot->packet_len;
memcpy(io->buf, slot->packet, slot->packet_len);
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) != 0) {
io->stats.packets++;
io->stats.bytes += slot->packet_len;
io->mbuf = NULL;
bbl_txq_read_next(txq);
if(burst) burst--;
} else {
/* Send traffic streams up to allowed burst. */
if(++stream_packets > io->stream_burst) {
break;
}
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
io->stats.io_errors++;
burst = 0;
break;
}
}
/* Get TX timestamp */
clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
while(burst) {
/* Send traffic streams up to allowed burst. */
if(!io->mbuf) {
if(!io_dpdk_mbuf_alloc(io)) {
break;
}
}
stream = bbl_stream_io_send_iter(io);
if(unlikely(stream == NULL)) {
break;
}
/* Transmit the packet. */
io->mbuf->data_len = stream->tx_len;
memcpy(io->buf, stream->tx_buf, stream->tx_len);
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) != 0) {
stream->tx_packets++;
stream->flow_seq++;
io->stats.packets++;
io->stats.bytes += stream->tx_len;
io->mbuf = NULL;
burst--;
} else {
io->stats.io_errors++;
burst = 0;
}
}
io->mbuf->data_len = io->buf_len;
/* Transmit the packet. */
if(rte_eth_tx_burst(interface->port_id, io->queue, &io->mbuf, 1) == 0) {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
return;
}
io->stats.packets++;
io->stats.bytes += io->buf_len;
io->mbuf = NULL;
io->buf = 0;
io->buf_len = 0;
}
}
bool
io_dpdk_add_mbuf_pool(io_handle_s *io)
{
@@ -475,11 +510,11 @@ io_dpdk_interface_init(bbl_interface_s *interface)
uint16_t id;
uint16_t nb_rx_queue = 1;
uint16_t nb_tx_queue = 1;
struct rte_eth_dev_info dev_info;
struct rte_eth_dev_info dev_info = {0};
struct rte_eth_conf local_port_conf = port_conf;
struct rte_eth_rxconf rx_conf;
struct rte_eth_txconf tx_conf;
struct rte_ether_addr mac;
struct rte_eth_rxconf rx_conf = {0};
struct rte_eth_txconf tx_conf = {0};
struct rte_ether_addr mac = {0};
io_handle_s *io;
RTE_ETH_FOREACH_DEV(port_id) {
@@ -523,7 +558,7 @@ io_dpdk_interface_init(bbl_interface_s *interface)
local_port_conf.rxmode.mq_mode = RTE_ETH_MQ_RX_RSS;
local_port_conf.rx_adv_conf.rss_conf.rss_hf =
(RTE_ETH_RSS_IP | RTE_ETH_RSS_TCP | RTE_ETH_RSS_UDP) &
(RTE_ETH_RSS_VLAN | RTE_ETH_RSS_IP | RTE_ETH_RSS_TCP | RTE_ETH_RSS_UDP | RTE_ETH_RSS_PPPOE | RTE_ETH_RSS_L2TPV2| RTE_ETH_RSS_MPLS) &
dev_info.flow_type_rss_offloads;
ret = rte_eth_dev_configure(port_id, nb_rx_queue, nb_tx_queue, &local_port_conf);
@@ -557,7 +592,6 @@ io_dpdk_interface_init(bbl_interface_s *interface)
io->next = interface->io.rx;
interface->io.rx = io;
io->interface = interface;
CIRCLEQ_INIT(&io->stream_tx_qhead);
if(config->rx_threads) {
if(!io_thread_init(io)) {
return false;
@@ -565,7 +599,7 @@ io_dpdk_interface_init(bbl_interface_s *interface)
io->thread->run_fn = io_dpdk_thread_rx_run_fn;
} else {
timer_add_periodic(&g_ctx->timer_root, &interface->io.rx_job, "RX", 0,
config->rx_interval, io, &io_dpdk_rx_job);
config->rx_interval, io, &io_dpdk_rx_job);
}
io->queue = queue;
if(!io_dpdk_add_mbuf_pool(io)) {
@@ -597,15 +631,11 @@ io_dpdk_interface_init(bbl_interface_s *interface)
interface->io.tx = io;
io->interface = interface;
io->buf = malloc(IO_BUFFER_LEN);
CIRCLEQ_INIT(&io->stream_tx_qhead);
if(config->tx_threads) {
if(!io_thread_init(io)) {
return false;
}
timer_add_periodic(&io->thread->timer.root, &io->thread->timer.io, "TX (threaded)", 0,
config->tx_interval, io->thread, &io_dpdk_thread_tx_job);
io->thread->timer.io->reset = false;
io->thread->set_cpu_affinity = true;
io->thread->run_fn = io_dpdk_thread_tx_run_fn;
} else {
timer_add_periodic(&g_ctx->timer_root, &interface->io.tx_job, "TX", 0,
config->tx_interval, io, &io_dpdk_tx_job);
code/bngblaster/src/io/io_interface.c
-2
View File
@@ -89,7 +89,6 @@ io_interface_init_rx(bbl_interface_s *interface)
io->next = interface->io.rx;
interface->io.rx = io;
io->interface = interface;
CIRCLEQ_INIT(&io->stream_tx_qhead);
if(config->rx_threads) {
if(!io_thread_init(io)) {
return false;
@@ -136,7 +135,6 @@ io_interface_init_tx(bbl_interface_s *interface)
io->next = interface->io.tx;
interface->io.tx = io;
io->interface = interface;
CIRCLEQ_INIT(&io->stream_tx_qhead);
if(config->tx_threads) {
if(!io_thread_init(io)) {
return false;
code/bngblaster/src/io/io_packet_mmap.c
+70 -59
View File
@@ -88,17 +88,25 @@ io_packet_mmap_rx_job(timer_s *timer)
//eth->timestamp.tv_nsec = tphdr->tp_nsec; /* ktime/hw timestamp */
eth->timestamp.tv_sec = io->timestamp.tv_sec;
eth->timestamp.tv_nsec = io->timestamp.tv_nsec;
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
bbl_rx_handler(interface, eth);
} else if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
}
/* Return ownership back to kernel */
tphdr->tp_status = TP_STATUS_KERNEL;
@@ -124,7 +132,9 @@ io_packet_mmap_tx_job(timer_s *timer)
struct tpacket2_hdr* tphdr;
uint8_t *frame_ptr;
uint32_t stream_packets = 0;
bbl_stream_s *stream = NULL;
uint16_t burst = interface->config->io_burst;
bool ctrl = true;
bool pcap = false;
@@ -132,8 +142,6 @@ io_packet_mmap_tx_job(timer_s *timer)
assert(io->direction == IO_EGRESS);
assert(io->thread == NULL);
io_update_stream_token_bucket(io);
frame_ptr = io->ring + (io->cursor * io->req.tp_frame_size);
tphdr = (struct tpacket2_hdr*)frame_ptr;
if(tphdr->tp_status != TP_STATUS_AVAILABLE) {
@@ -145,7 +153,7 @@ io_packet_mmap_tx_job(timer_s *timer)
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
while(true) {
while(burst) {
/* Check if this slot available for writing. */
if(tphdr->tp_status != TP_STATUS_AVAILABLE) {
io->stats.no_buffer++;
@@ -160,21 +168,20 @@ io_packet_mmap_tx_job(timer_s *timer)
continue;
}
} else {
/* Send traffic streams up to allowed burst. */
if(++stream_packets > io->stream_burst) {
stream = bbl_stream_io_send_iter(io);
if(!stream) {
break;
}
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
break;
}
}
memcpy(io->buf, stream->tx_buf, stream->tx_len);
io->buf_len = stream->tx_len;
}
tphdr->tp_len = io->buf_len;
tphdr->tp_status = TP_STATUS_SEND_REQUEST;
io->queued++;
io->stats.packets++;
io->stats.bytes += io->buf_len;
burst--;
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf && (ctrl || g_ctx->pcap.include_streams)) {
@@ -225,15 +232,14 @@ io_packet_mmap_thread_rx_run_fn(io_thread_s *thread)
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 0;
sleep.tv_nsec = 10000; /* 0.01ms */
while(thread->active) {
frame_ptr = ring + (cursor * frame_size);
tphdr = (struct tpacket2_hdr*)frame_ptr;
if(!(tphdr->tp_status & TP_STATUS_USER)) {
/* If no buffer is available poll kernel */
poll_kernel(io, POLLIN);
sleep.tv_nsec = 100000; /* 0.1ms */
//poll_kernel(io, POLLIN);
nanosleep(&sleep, &rem);
continue;
}
@@ -254,18 +260,13 @@ io_packet_mmap_thread_rx_run_fn(io_thread_s *thread)
frame_ptr = ring + (cursor * frame_size);
tphdr = (struct tpacket2_hdr*)frame_ptr;
}
sleep.tv_nsec = 1000; /* 0.001ms */
nanosleep(&sleep, &rem);
}
}
/**
* This job is for PACKET_MMAP TX in worker thread!
*/
void
io_packet_mmap_thread_tx_job(timer_s *timer)
io_packet_mmap_thread_tx_run_fn(io_thread_s *thread)
{
io_thread_s *thread = timer->data;
io_handle_s *io = thread->io;
bbl_interface_s *interface = io->interface;
@@ -275,30 +276,41 @@ io_packet_mmap_thread_tx_job(timer_s *timer)
struct tpacket2_hdr* tphdr;
uint8_t *frame_ptr;
uint32_t stream_packets = 0;
bbl_stream_s *stream = NULL;
uint16_t io_burst = interface->config->io_burst;
uint16_t burst = 0;
bool ctrl = true;
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 10;
assert(io->mode == IO_MODE_PACKET_MMAP);
assert(io->direction == IO_EGRESS);
assert(io->thread);
io_update_stream_token_bucket(io);
while(thread->active) {
nanosleep(&sleep, &rem);
frame_ptr = io->ring + (io->cursor * io->req.tp_frame_size);
tphdr = (struct tpacket2_hdr *)frame_ptr;
if(tphdr->tp_status != TP_STATUS_AVAILABLE) {
/* If no buffer is available poll kernel. */
io->stats.no_buffer++;
poll_kernel(io, POLLOUT);
continue;
}
frame_ptr = io->ring + (io->cursor * io->req.tp_frame_size);
tphdr = (struct tpacket2_hdr *)frame_ptr;
if(tphdr->tp_status != TP_STATUS_AVAILABLE) {
/* If no buffer is available poll kernel. */
poll_kernel(io, POLLOUT);
io->stats.no_buffer++;
} else {
/* Get TX timestamp */
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
while(true) {
/* Check if this slot available for writing. */
clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
burst = io_burst;
ctrl = true;
while(burst) {
if(tphdr->tp_status != TP_STATUS_AVAILABLE) {
io->stats.no_buffer++;
poll_kernel(io, POLLOUT);
break;
}
io->buf = frame_ptr + TPACKET2_HDRLEN - sizeof(struct sockaddr_ll);
@@ -316,12 +328,12 @@ io_packet_mmap_thread_tx_job(timer_s *timer)
}
} else {
/* Send traffic streams up to allowed burst. */
if(++stream_packets > io->stream_burst) {
break;
}
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
stream = bbl_stream_io_send_iter(io);
if(!stream) {
break;
}
memcpy(io->buf, stream->tx_buf, stream->tx_len);
io->buf_len = stream->tx_len;
}
tphdr->tp_len = io->buf_len;
@@ -330,22 +342,23 @@ io_packet_mmap_thread_tx_job(timer_s *timer)
io->queued++;
io->stats.packets++;
io->stats.bytes += io->buf_len;
burst--;
/* Get next slot. */
io->cursor = (io->cursor + 1) % io->req.tp_frame_nr;
frame_ptr = io->ring + (io->cursor * io->req.tp_frame_size);
tphdr = (struct tpacket2_hdr *)frame_ptr;
}
}
if(io->queued) {
/* Notify kernel. */
if(sendto(io->fd, NULL, 0, 0, NULL, 0) < 0) {
LOG(IO, "PACKET_MMAP sendto on interface %s failed with error %s (%d)\n",
interface->name, strerror(errno), errno);
io->stats.io_errors++;
} else {
io->queued = 0;
if(io->queued) {
/* Notify kernel. */
if(sendto(io->fd, NULL, 0, 0, NULL, 0) < 0) {
LOG(IO, "PACKET_MMAP sendto on interface %s failed with error %s (%d)\n",
interface->name, strerror(errno), errno);
io->stats.io_errors++;
} else {
io->queued = 0;
}
}
}
}
@@ -366,9 +379,7 @@ io_packet_mmap_init(io_handle_s *io)
if(io->direction == IO_INGRESS) {
thread->run_fn = io_packet_mmap_thread_rx_run_fn;
} else {
timer_add_periodic(&thread->timer.root, &thread->timer.io, "TX (threaded)", 0,
config->tx_interval, thread, &io_packet_mmap_thread_tx_job);
thread->timer.io->reset = false;
thread->run_fn = io_packet_mmap_thread_tx_run_fn;
}
} else {
if(io->direction == IO_INGRESS) {
code/bngblaster/src/io/io_raw.c
+124 -108
View File
@@ -45,17 +45,25 @@ io_raw_rx_job(timer_s *timer)
/* Copy RX timestamp */
eth->timestamp.tv_sec = io->timestamp.tv_sec;
eth->timestamp.tv_nsec = io->timestamp.tv_nsec;
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
bbl_rx_handler(interface, eth);
} else if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
/* Dump the packet into pcap file */
if(g_ctx->pcap.write_buf) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
}
}
if(pcap) {
@@ -89,71 +97,85 @@ io_raw_tx_job(timer_s *timer)
io_handle_s *io = timer->data;
bbl_interface_s *interface = io->interface;
uint32_t stream_packets = 0;
bool ctrl = true;
bbl_stream_s *stream = NULL;
uint16_t burst = interface->config->io_burst;
bool pcap = false;
assert(io->mode == IO_MODE_RAW);
assert(io->direction == IO_EGRESS);
assert(io->thread == NULL);
io_update_stream_token_bucket(io);
/* Get TX timestamp */
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
while(true) {
/* If sendto fails, the failed packet remains in TX buffer to be retried
* in the next interval. */
if(io->buf_len) {
if(packet_is_bbl(io->buf, io->buf_len)) {
/* Update timestamp if BBL traffic is retried. */
*(uint32_t*)(io->buf + (io->buf_len - 8)) = io->timestamp.tv_sec;
*(uint32_t*)(io->buf + (io->buf_len - 4)) = io->timestamp.tv_nsec;
}
} else if(ctrl) {
/* First send all control traffic which has higher priority. */
while(burst) {
if(likely(io->buf_len == 0)) {
if(bbl_tx(interface, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
io->buf_len = 0;
ctrl = false;
continue;
}
} else {
/* Send traffic streams up to allowed burst. */
if(++stream_packets > io->stream_burst) {
break;
}
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
break;
}
}
if(sendto(io->fd, io->buf, io->buf_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) <0 ) {
if(errno == EMSGSIZE) {
io_raw_tx_lo_long(io);
} else {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
interface->name, strerror(errno), errno);
io->stats.io_errors++;
break;
}
} else {
if(sendto(io->fd, io->buf, io->buf_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) > 0) {
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf && (ctrl || g_ctx->pcap.include_streams)) {
if(unlikely(g_ctx->pcap.write_buf != NULL)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, io->buf, io->buf_len,
interface->ifindex, PCAPNG_EPB_FLAGS_OUTBOUND);
}
io->stats.packets++;
io->stats.bytes += io->buf_len;
io->buf_len = 0;
burst--;
} else {
if(errno == EMSGSIZE) {
io_raw_tx_lo_long(io);
io->buf_len = 0;
} else {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
interface->name, strerror(errno), errno);
io->stats.io_errors++;
burst = 0;
}
}
io->buf_len = 0;
}
if(pcap) {
while(burst) {
/* Send traffic streams up to allowed burst. */
stream = bbl_stream_io_send_iter(io);
if(unlikely(stream == NULL)) {
break;
}
if(sendto(io->fd, stream->tx_buf, stream->tx_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) > 0) {
/* Dump the packet into pcap file. */
if(unlikely(g_ctx->pcap.write_buf && g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&io->timestamp, stream->tx_buf, stream->tx_len,
interface->ifindex, PCAPNG_EPB_FLAGS_OUTBOUND);
}
stream->tx_packets++;
stream->flow_seq++;
io->stats.packets++;
io->stats.bytes += stream->tx_len;
burst--;
} else {
if(errno == EMSGSIZE) {
io->buf_len = stream->tx_len;
io_raw_tx_lo_long(io);
io->buf_len = 0;
} else {
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
interface->name, strerror(errno), errno);
io->stats.io_errors++;
burst = 0;
}
}
}
if(unlikely(pcap)) {
pcapng_fflush();
}
}
@@ -166,12 +188,11 @@ io_raw_thread_rx_run_fn(io_thread_s *thread)
struct sockaddr saddr;
int saddr_size = sizeof(saddr);
struct timespec sleep, rem;
assert(io->direction == IO_INGRESS);
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 0;
sleep.tv_nsec = 1000; /* 0.001ms */
while(thread->active) {
/* Get RX timestamp */
@@ -179,7 +200,6 @@ io_raw_thread_rx_run_fn(io_thread_s *thread)
/* Receive from socket */
io->buf_len = recvfrom(io->fd, io->buf, IO_BUFFER_LEN, 0, &saddr , (socklen_t*)&saddr_size);
if(io->buf_len < 14 || io->buf_len > IO_BUFFER_LEN) {
sleep.tv_nsec = 1000; /* 0.001ms */
nanosleep(&sleep, &rem);
continue;
}
@@ -188,77 +208,75 @@ io_raw_thread_rx_run_fn(io_thread_s *thread)
}
}
/**
* This job is for RAW TX in worker thread!
*/
void
io_raw_thread_tx_job(timer_s *timer)
io_raw_thread_tx_run_fn(io_thread_s *thread)
{
io_thread_s *thread = timer->data;
io_handle_s *io = thread->io;
bbl_interface_s *interface = io->interface;
bbl_txq_s *txq = thread->txq;
bbl_txq_slot_t *slot;
uint32_t stream_packets = 0;
bbl_stream_s *stream = NULL;
uint16_t io_burst = interface->config->io_burst;
uint16_t burst = 0;
struct timespec sleep, rem;
sleep.tv_sec = 0;
sleep.tv_nsec = 1000 * io_burst;
assert(io->mode == IO_MODE_RAW);
assert(io->direction == IO_EGRESS);
assert(io->thread);
io_update_stream_token_bucket(io);
while(thread->active) {
nanosleep(&sleep, &rem);
burst = io_burst;
/* First send all control traffic which has higher priority. */
while((slot = bbl_txq_read_slot(txq))) {
/* This packet will be retried next interval
* because slot is not marked as read. */
if(sendto(io->fd, slot->packet, slot->packet_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) <0 ) {
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
io->interface->name, strerror(errno), errno);
io->stats.io_errors++;
return;
}
io->stats.packets++;
io->stats.bytes += slot->packet_len;
bbl_txq_read_next(txq);
}
/* Get TX timestamp */
//clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
io->timestamp.tv_sec = timer->timestamp->tv_sec;
io->timestamp.tv_nsec = timer->timestamp->tv_nsec;
/* Send traffic streams up to allowed burst. */
while(stream_packets++ < io->stream_burst) {
/* If sendto fails, the failed packet remains in TX buffer
* to be retried in the next interval. */
if(io->buf_len) {
if(packet_is_bbl(io->buf, io->buf_len)) {
/* Update timestamp if BBL traffic is retried. */
*(uint32_t*)(io->buf + (io->buf_len - 8)) = io->timestamp.tv_sec;
*(uint32_t*)(io->buf + (io->buf_len - 4)) = io->timestamp.tv_nsec;
}
} else {
if(bbl_stream_tx(io, io->buf, &io->buf_len) != PROTOCOL_SUCCESS) {
break;
}
}
if(sendto(io->fd, io->buf, io->buf_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) <0 ) {
if(errno == EMSGSIZE) {
io_raw_tx_lo_long(io);
/* First send all control traffic which has higher priority. */
while((slot = bbl_txq_read_slot(txq))) {
if(sendto(io->fd, slot->packet, slot->packet_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) >= 0 ) {
io->stats.packets++;
io->stats.bytes += slot->packet_len;
bbl_txq_read_next(txq);
if(burst) burst--;
} else {
/* This packet will be retried next interval
* because io->buf_len is not reset to zero. */
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
io->interface->name, strerror(errno), errno);
io->stats.io_errors++;
burst = 0;
break;
}
} else {
io->stats.packets++;
io->stats.bytes += io->buf_len;
}
io->buf_len = 0;
/* Get TX timestamp */
clock_gettime(CLOCK_MONOTONIC, &io->timestamp);
while(burst) {
/* Send traffic streams up to allowed burst. */
stream = bbl_stream_io_send_iter(io);
if(unlikely(stream == NULL)) {
break;
}
if(unlikely(sendto(io->fd, stream->tx_buf, stream->tx_len, 0, (struct sockaddr*)&io->addr, sizeof(struct sockaddr_ll)) >=0)) {
stream->tx_packets++;
stream->flow_seq++;
io->stats.packets++;
io->stats.bytes += stream->tx_len;
burst--;
} else {
if(errno == EMSGSIZE) {
io->buf_len = stream->tx_len;
io_raw_tx_lo_long(io);
io->buf_len = 0;
} else {
LOG(IO, "RAW sendto on interface %s failed with error %s (%d)\n",
io->interface->name, strerror(errno), errno);
io->stats.io_errors++;
burst = 0;
}
}
}
}
}
@@ -280,9 +298,7 @@ io_raw_init(io_handle_s *io)
if(io->direction == IO_INGRESS) {
thread->run_fn = io_raw_thread_rx_run_fn;
} else {
timer_add_periodic(&thread->timer.root, &thread->timer.io, "TX (threaded)", 0,
config->tx_interval, thread, &io_raw_thread_tx_job);
thread->timer.io->reset = false;
thread->run_fn = io_raw_thread_tx_run_fn;
}
} else {
if(io->direction == IO_INGRESS) {
code/bngblaster/src/io/io_thread.c
+19 -26
View File
@@ -62,7 +62,7 @@ io_thread_rx_handler(io_thread_s *thread, io_handle_s *io)
io->stats.packets++;
io->stats.bytes += io->buf_len;
if(likely(packet_is_bbl(io->buf, io->buf_len))) {
if(packet_is_bbl(io->buf, io->buf_len)) {
/** Process */
decode_result = decode_ethernet(io->buf, io->buf_len, thread->sp, SCRATCHPAD_LEN, &eth);
if(decode_result == PROTOCOL_SUCCESS) {
@@ -132,17 +132,25 @@ io_thread_main_rx_job(timer_s *timer)
/* Copy RX timestamp */
eth->timestamp.tv_sec = slot->timestamp.tv_sec;
eth->timestamp.tv_nsec = slot->timestamp.tv_nsec;
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&slot->timestamp, slot->packet, slot->packet_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
bbl_rx_handler(interface, eth);
} else if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf && (!eth->bbl || g_ctx->pcap.include_streams)) {
pcap = true;
pcapng_push_packet_header(&slot->timestamp, slot->packet, slot->packet_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
/* Dump the packet into pcap file. */
if(g_ctx->pcap.write_buf) {
pcap = true;
pcapng_push_packet_header(&slot->timestamp, slot->packet, slot->packet_len,
interface->ifindex, PCAPNG_EPB_FLAGS_INBOUND);
}
if(decode_result == UNKNOWN_PROTOCOL) {
io->stats.unknown++;
} else {
io->stats.protocol_errors++;
}
}
bbl_txq_read_next(thread->txq);
}
@@ -211,17 +219,6 @@ io_thread_main(void *thread_data)
return NULL;
}
void
io_thread_timer_loop(io_thread_s *thread)
{
pthread_mutex_lock(&thread->mutex);
timer_smear_all_buckets(&thread->timer.root);
pthread_mutex_unlock(&thread->mutex);
while(thread->active) {
timer_walk(&thread->timer.root);
}
}
bool
io_thread_init(io_handle_s *io)
{
@@ -262,11 +259,8 @@ io_thread_init(io_handle_s *io)
return false;
}
/* Init thread timer root */
timer_init_root(&thread->timer.root);
/* Default run function which might be overwritten */
thread->run_fn = io_thread_timer_loop;
thread->run_fn = NULL;
/* Add thread main loop timers/jobs */
if(io->direction == IO_INGRESS && !interface->io.rx_job) {
@@ -313,7 +307,6 @@ io_thread_start_all()
sleep.tv_nsec = 7800179; /* prime number between 7 and 8ms */
while(thread) {
thread->active = true;
timer_smear_all_buckets(&thread->timer.root);
pthread_create(&thread->thread, NULL, io_thread_main, (void *)thread);
if(thread->set_cpu_affinity) {
if(pthread_setaffinity_np(thread->thread, sizeof(cpu_set_t), &thread->cpuset)) {