diff options
Diffstat (limited to 'VNFs/DPPD-PROX/handle_genl4.c')
| -rw-r--r-- | VNFs/DPPD-PROX/handle_genl4.c | 1139 |
1 files changed, 1139 insertions, 0 deletions
diff --git a/VNFs/DPPD-PROX/handle_genl4.c b/VNFs/DPPD-PROX/handle_genl4.c new file mode 100644 index 00000000..4c62c641 --- /dev/null +++ b/VNFs/DPPD-PROX/handle_genl4.c @@ -0,0 +1,1139 @@ +/* +// Copyright (c) 2010-2017 Intel Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +*/ + +#include <rte_mbuf.h> +#include <pcap.h> +#include <string.h> +#include <stdlib.h> +#include <rte_cycles.h> +#include <rte_version.h> +#include <rte_ip.h> +#include <rte_udp.h> +#include <rte_tcp.h> +#include <rte_hash.h> +#include <rte_hash_crc.h> + +#include "prox_lua.h" +#include "prox_lua_types.h" +#include "prox_malloc.h" +#include "file_utils.h" +#include "hash_set.h" +#include "prox_assert.h" +#include "prox_args.h" +#include "defines.h" +#include "pkt_parser.h" +#include "handle_lat.h" +#include "task_init.h" +#include "task_base.h" +#include "prox_port_cfg.h" +#include "lconf.h" +#include "log.h" +#include "quit.h" +#include "heap.h" +#include "mbuf_utils.h" +#include "genl4_bundle.h" +#include "genl4_stream_udp.h" +#include "genl4_stream_tcp.h" +#include "cdf.h" +#include "fqueue.h" +#include "token_time.h" +#include "commands.h" +#include "prox_shared.h" + +#if RTE_VERSION < RTE_VERSION_NUM(1,8,0,0) +#define RTE_CACHE_LINE_SIZE CACHE_LINE_SIZE +#endif + +struct new_tuple { + uint32_t dst_addr; + uint8_t proto_id; + uint16_t dst_port; + uint16_t l2_types[4]; +} __attribute__((packed)); + +enum handle_state {HANDLE_QUEUED, HANDLE_SCHEDULED}; + +struct task_gen_server { + struct task_base base; + struct l4_stats l4_stats; + struct rte_mempool *mempool; + struct rte_hash *listen_hash; + /* Listening bundles contain only 1 part since the state of a + multi_part comm is kept mostly at the client side*/ + struct bundle_cfg **listen_entries; + struct bundle_ctx_pool bundle_ctx_pool; + struct bundle_cfg *bundle_cfgs; /* Loaded configurations */ + struct token_time token_time; + enum handle_state handle_state; + struct heap *heap; + struct fqueue *fqueue; + struct rte_mbuf *cur_mbufs[MAX_PKT_BURST]; + uint32_t cur_mbufs_beg; + uint32_t cur_mbufs_end; + uint32_t cancelled; + uint8_t out_saved; + struct rte_mbuf *mbuf_saved; + uint64_t last_tsc; + unsigned seed; + /* Handle scheduled events */ + struct rte_mbuf *new_mbufs[MAX_PKT_BURST]; + uint32_t n_new_mbufs; +}; + +struct task_gen_client { + struct task_base base; + struct l4_stats l4_stats; + struct rte_mempool *mempool; + struct bundle_ctx_pool bundle_ctx_pool; + struct bundle_cfg *bundle_cfgs; /* Loaded configurations */ + struct token_time token_time; + /* Create new connections and handle scheduled events */ + struct rte_mbuf *new_mbufs[MAX_PKT_BURST]; + uint32_t new_conn_cost; + uint32_t new_conn_tokens; + uint64_t new_conn_last_tsc; + uint32_t n_new_mbufs; + uint64_t last_tsc; + struct cdf *cdf; + unsigned seed; + struct heap *heap; +}; + +static int refill_mbufs(uint32_t *n_new_mbufs, struct rte_mempool *mempool, struct rte_mbuf **mbufs) +{ + if (*n_new_mbufs == MAX_PKT_BURST) + return 0; + + if (rte_mempool_get_bulk(mempool, (void **)mbufs, MAX_PKT_BURST - *n_new_mbufs) < 0) { + plogx_err("4Mempool alloc failed for %d mbufs\n", MAX_PKT_BURST - *n_new_mbufs); + return -1; + } + + for (uint32_t i = 0; i < MAX_PKT_BURST - *n_new_mbufs; ++i) { + init_mbuf_seg(mbufs[i]); + } + + *n_new_mbufs = MAX_PKT_BURST; + + return 0; +} + +static const struct bundle_cfg *server_accept(struct task_gen_server *task, struct new_tuple *nt) +{ + int ret = rte_hash_lookup(task->listen_hash, nt); + + if (ret < 0) + return NULL; + else + return task->listen_entries[ret]; +} + +static int handle_gen_bulk_client(struct task_base *tbase, struct rte_mbuf **mbufs, uint16_t n_pkts) +{ + struct task_gen_client *task = (struct task_gen_client *)tbase; + uint8_t out[MAX_PKT_BURST] = {0}; + struct bundle_ctx *conn; + int ret; + + if (n_pkts) { + for (int i = 0; i < n_pkts; ++i) { + struct pkt_tuple pt; + struct l4_meta l4_meta; + + if (parse_pkt(mbufs[i], &pt, &l4_meta)) { + plogdx_err(mbufs[i], "Parsing failed\n"); + out[i] = OUT_DISCARD; + continue; + } + + ret = rte_hash_lookup(task->bundle_ctx_pool.hash, (const void *)&pt); + + if (ret < 0) { + plogx_dbg("Client: packet RX that does not belong to connection:" + "Client = "IPv4_BYTES_FMT":%d, Server = "IPv4_BYTES_FMT":%d\n", + IPv4_BYTES(((uint8_t*)&pt.dst_addr)), + rte_bswap16(pt.dst_port), + IPv4_BYTES(((uint8_t*)&pt.src_addr)), + rte_bswap16(pt.src_port)); + + plogdx_dbg(mbufs[i], NULL); + + if (pt.proto_id == IPPROTO_TCP) { + stream_tcp_create_rst(mbufs[i], &l4_meta, &pt); + out[i] = 0; + continue; + } + else { + out[i] = OUT_DISCARD; + continue; + } + } + + conn = task->bundle_ctx_pool.hash_entries[ret]; + ret = bundle_proc_data(conn, mbufs[i], &l4_meta, &task->bundle_ctx_pool, &task->seed, &task->l4_stats); + out[i] = ret == 0? 0: OUT_HANDLED; + } + task->base.tx_pkt(&task->base, mbufs, n_pkts, out); + } + + /* If there is at least one callback to handle, handle at most MAX_PKT_BURST */ + if (heap_top_is_lower(task->heap, rte_rdtsc())) { + if (0 != refill_mbufs(&task->n_new_mbufs, task->mempool, task->new_mbufs)) + return 0; + + uint16_t n_called_back = 0; + while (heap_top_is_lower(task->heap, rte_rdtsc()) && n_called_back < MAX_PKT_BURST) { + conn = BUNDLE_CTX_UPCAST(heap_pop(task->heap)); + + /* handle packet TX (retransmit or delayed transmit) */ + ret = bundle_proc_data(conn, task->new_mbufs[n_called_back], NULL, &task->bundle_ctx_pool, &task->seed, &task->l4_stats); + + if (ret == 0) { + out[n_called_back] = 0; + n_called_back++; + } + } + plogx_dbg("During callback, will send %d packets\n", n_called_back); + + task->base.tx_pkt(&task->base, task->new_mbufs, n_called_back, out); + task->n_new_mbufs -= n_called_back; + } + + uint32_t n_new = task->bundle_ctx_pool.n_free_bundles; + n_new = n_new > MAX_PKT_BURST? MAX_PKT_BURST : n_new; + + uint64_t diff = (rte_rdtsc() - task->new_conn_last_tsc)/task->new_conn_cost; + task->new_conn_last_tsc += diff * task->new_conn_cost; + task->new_conn_tokens += diff; + + if (task->new_conn_tokens > 16) + task->new_conn_tokens = 16; + if (n_new > task->new_conn_tokens) + n_new = task->new_conn_tokens; + task->new_conn_tokens -= n_new; + if (n_new == 0) + return 0; + + if (0 != refill_mbufs(&task->n_new_mbufs, task->mempool, task->new_mbufs)) + return 0; + + for (uint32_t i = 0; i < n_new; ++i) { + struct bundle_ctx *bundle_ctx = bundle_ctx_pool_get_w_cfg(&task->bundle_ctx_pool); + PROX_ASSERT(bundle_ctx); + + struct pkt_tuple *pt = &bundle_ctx->tuple; + + int n_retries = 0; + do { + /* Note that the actual packet sent will + contain swapped addresses and ports + (i.e. pkt.src <=> tuple.dst). The incoming + packet will match this struct. */ + bundle_init(bundle_ctx, task->heap, PEER_CLIENT, &task->seed); + + ret = rte_hash_lookup(task->bundle_ctx_pool.hash, (const void *)pt); + if (ret >= 0) { + if (n_retries++ == 1000) { + plogx_err("Already tried 1K times\n"); + } + } + } while (ret >= 0); + + ret = rte_hash_add_key(task->bundle_ctx_pool.hash, (const void *)pt); + + if (ret < 0) { + plogx_err("Failed to add key ret = %d, n_free = %d\n", ret, task->bundle_ctx_pool.n_free_bundles); + bundle_ctx_pool_put(&task->bundle_ctx_pool, bundle_ctx); + + pkt_tuple_debug2(pt); + out[i] = OUT_DISCARD; + continue; + } + + task->bundle_ctx_pool.hash_entries[ret] = bundle_ctx; + + if (bundle_ctx->ctx.stream_cfg->proto == IPPROTO_TCP) + task->l4_stats.tcp_created++; + else + task->l4_stats.udp_created++; + + task->l4_stats.bundles_created++; + + ret = bundle_proc_data(bundle_ctx, task->new_mbufs[i], NULL, &task->bundle_ctx_pool, &task->seed, &task->l4_stats); + out[i] = ret == 0? 0: OUT_HANDLED; + } + + int ret2 = task->base.tx_pkt(&task->base, task->new_mbufs, n_new, out); + task->n_new_mbufs -= n_new; + return ret2; +} + +static int handle_gen_queued(struct task_gen_server *task) +{ + uint8_t out[MAX_PKT_BURST]; + struct bundle_ctx *conn; + struct pkt_tuple pkt_tuple; + struct l4_meta l4_meta; + uint16_t j; + uint16_t cancelled = 0; + int ret; + + if (task->cur_mbufs_beg == task->cur_mbufs_end) { + task->cur_mbufs_end = fqueue_get(task->fqueue, task->cur_mbufs, MAX_PKT_BURST); + task->cur_mbufs_beg = 0; + } + uint16_t n_pkts = task->cur_mbufs_end - task->cur_mbufs_beg; + struct rte_mbuf **mbufs = task->cur_mbufs + task->cur_mbufs_beg; + + j = task->cancelled; + if (task->cancelled) { + uint16_t pkt_len = mbuf_wire_size(mbufs[0]); + + if (token_time_take(&task->token_time, pkt_len) != 0) + return -1; + + out[0] = task->out_saved; + task->cancelled = 0; + } + + /* Main proc loop */ + for (; j < n_pkts; ++j) { + + if (parse_pkt(mbufs[j], &pkt_tuple, &l4_meta)) { + plogdx_err(mbufs[j], "Unknown packet, parsing failed\n"); + out[j] = OUT_DISCARD; + } + + conn = NULL; + ret = rte_hash_lookup(task->bundle_ctx_pool.hash, (const void *)&pkt_tuple); + + if (ret >= 0) + conn = task->bundle_ctx_pool.hash_entries[ret]; + else { + /* If not part of existing connection, try to create a connection */ + struct new_tuple nt; + nt.dst_addr = pkt_tuple.dst_addr; + nt.proto_id = pkt_tuple.proto_id; + nt.dst_port = pkt_tuple.dst_port; + rte_memcpy(nt.l2_types, pkt_tuple.l2_types, sizeof(nt.l2_types)); + const struct bundle_cfg *n; + + if (NULL != (n = server_accept(task, &nt))) { + conn = bundle_ctx_pool_get(&task->bundle_ctx_pool); + if (!conn) { + out[j] = OUT_DISCARD; + plogx_err("No more free bundles to accept new connection\n"); + continue; + } + ret = rte_hash_add_key(task->bundle_ctx_pool.hash, (const void *)&pkt_tuple); + if (ret < 0) { + out[j] = OUT_DISCARD; + bundle_ctx_pool_put(&task->bundle_ctx_pool, conn); + plog_err("Adding key failed while trying to accept connection\n"); + continue; + } + + task->bundle_ctx_pool.hash_entries[ret] = conn; + + bundle_init_w_cfg(conn, n, task->heap, PEER_SERVER, &task->seed); + conn->tuple = pkt_tuple; + + if (conn->ctx.stream_cfg->proto == IPPROTO_TCP) + task->l4_stats.tcp_created++; + else + task->l4_stats.udp_created++; + } + else { + plog_err("Packet received for service that does not exist :\n" + "source ip = %0x:%u\n" + "dst ip = %0x:%u\n", + pkt_tuple.src_addr, rte_bswap16(pkt_tuple.src_port), + pkt_tuple.dst_addr, rte_bswap16(pkt_tuple.dst_port)); + } + } + + /* bundle contains either an active connection or a + newly created connection. If it is NULL, then not + listening. */ + if (NULL != conn) { + ret = bundle_proc_data(conn, mbufs[j], &l4_meta, &task->bundle_ctx_pool, &task->seed, &task->l4_stats); + + out[j] = ret == 0? 0: OUT_HANDLED; + + if (ret == 0) { + uint16_t pkt_len = mbuf_wire_size(mbufs[j]); + + if (token_time_take(&task->token_time, pkt_len) != 0) { + task->out_saved = out[j]; + task->cancelled = 1; + task->base.tx_pkt(&task->base, mbufs, j, out); + task->cur_mbufs_beg += j; + return -1; + } + } + } + else { + pkt_tuple_debug(&pkt_tuple); + plogd_dbg(mbufs[j], NULL); + out[j] = OUT_DISCARD; + } + } + + task->base.tx_pkt(&task->base, mbufs, j, out); + + task->cur_mbufs_beg += j; + return 0; +} + +static int handle_gen_scheduled(struct task_gen_server *task) +{ + struct bundle_ctx *conn; + uint8_t out[MAX_PKT_BURST]; + int ret; + uint16_t n_called_back = 0; + + if (task->cancelled) { + struct rte_mbuf *mbuf = task->mbuf_saved; + + uint16_t pkt_len = mbuf_wire_size(mbuf); + if (token_time_take(&task->token_time, pkt_len) == 0) { + task->cancelled = 0; + out[0] = 0; + task->base.tx_pkt(&task->base, &mbuf, 1, out); + } + else { + return -1; + } + } + + if (0 != refill_mbufs(&task->n_new_mbufs, task->mempool, task->new_mbufs)) + return -1; + + conn = NULL; + while (heap_top_is_lower(task->heap, rte_rdtsc()) && n_called_back < task->n_new_mbufs) { + conn = BUNDLE_CTX_UPCAST(heap_pop(task->heap)); + + /* handle packet TX (retransmit or delayed transmit) */ + ret = bundle_proc_data(conn, task->new_mbufs[n_called_back], NULL, &task->bundle_ctx_pool, &task->seed, &task->l4_stats); + + if (ret == 0) { + struct rte_mbuf *mbuf = task->new_mbufs[n_called_back]; + uint16_t pkt_len = mbuf_wire_size(mbuf); + + if (token_time_take(&task->token_time, pkt_len) == 0) { + out[n_called_back] = 0; + n_called_back++; + } + else { + + struct ether_hdr *eth = rte_pktmbuf_mtod(mbuf, struct ether_hdr *); + struct ipv4_hdr *ip = (struct ipv4_hdr*)(eth + 1); + struct tcp_hdr *tcp = (struct tcp_hdr*)(ip + 1); + + task->out_saved = 0; + task->cancelled = 1; + task->mbuf_saved = mbuf; + task->base.tx_pkt(&task->base, task->new_mbufs, n_called_back, out); + /* The mbuf that is currently been + processed (and which has been + cancelled) is saved in + task->mbuf_saved. It will be + restored as the first mbuf when + this function is called again. */ + task->n_new_mbufs -= (n_called_back + 1); + return -1; + } + } + } + + task->base.tx_pkt(&task->base, task->new_mbufs, n_called_back, out); + task->n_new_mbufs -= n_called_back; + + return 0; +} + +static int handle_gen_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uint16_t n_pkts) +{ + struct task_gen_server *task = (struct task_gen_server *)tbase; + struct bundle_ctx *conn; + int ret, ret2 = 0; + + token_time_update(&task->token_time, rte_rdtsc()); + + if ((ret = fqueue_put(task->fqueue, mbufs, n_pkts)) != n_pkts) { + uint8_t out[MAX_PKT_BURST]; + for (uint16_t j = 0; j < n_pkts - ret; ++j) + out[j] = OUT_DISCARD; + + ret2 = task->base.tx_pkt(&task->base, mbufs + ret, n_pkts - ret, out); + } + if (task->handle_state == HANDLE_QUEUED) { + if (handle_gen_queued(task) == 0) { + if (handle_gen_scheduled(task) != 0) + task->handle_state = HANDLE_SCHEDULED; + } + } + else { + if (handle_gen_scheduled(task) == 0) { + if (handle_gen_queued(task) != 0) + task->handle_state = HANDLE_QUEUED; + } + } + return ret2; +} + +static int lua_to_host_set(struct lua_State *L, enum lua_place from, const char *name, struct host_set *h) +{ + int pop; + if ((pop = lua_getfrom(L, from, name)) < 0) + return -1; + + if (!lua_istable(L, -1)) + return -1; + + uint32_t port = 0, port_mask = 0; + + if (lua_to_ip(L, TABLE, "ip", &h->ip) || lua_to_int(L, TABLE, "port", &port)) + return -1; + + if (lua_to_int(L, TABLE, "ip_mask", &h->ip_mask)) + h->ip_mask = 0; + if (lua_to_int(L, TABLE, "port_mask", &port_mask)) + h->port_mask = 0; + + h->port = rte_bswap16(port); + h->port_mask = rte_bswap16(port_mask); + h->ip = rte_bswap32(h->ip); + h->ip_mask = rte_bswap32(h->ip_mask); + + lua_pop(L, pop); + return 0; +} + +static int file_read_cached(const char *file_name, uint8_t **mem, uint32_t beg, uint32_t len, uint32_t socket, struct hash_set *hs) +{ + if (len == 0) { + *mem = 0; + return 0; + } + + uint8_t *data_mem; + + /* Since the configuration can reference the same file from + multiple places, use prox_shared infrastructure to detect + this and return previously loaded data. */ + char name[256]; + + snprintf(name, sizeof(name), "%u-%u:%s", beg, len, file_name); + *mem = prox_sh_find_socket(socket, name); + if (*mem) + return 0; + + /* check if the file has been loaded on the other socket. */ + if (socket == 1 && (data_mem = prox_sh_find_socket(0, name))) { + uint8_t *data_find = hash_set_find(hs, data_mem, len); + if (!data_find) { + data_find = prox_zmalloc(len, socket); + PROX_PANIC(data_find == NULL, "Failed to allocate memory (%u bytes) to hold header for peer\n", len); + + rte_memcpy(data_find, data_mem, len); + hash_set_add(hs, data_find, len); + } + *mem = data_find; + prox_sh_add_socket(socket, name, *mem); + return 0; + } + + /* It is possible that a file with a different name contains + the same data. In that case, search all loaded files and + compare the data to reduce memory utilization.*/ + data_mem = malloc(len); + PROX_PANIC(data_mem == NULL, "Failed to allocate temporary memory to hold data\n"); + + if (file_read_content(file_name, data_mem, beg, len)) { + plog_err("%s\n", file_get_error()); + return -1; + } + + uint8_t *data_find = hash_set_find(hs, data_mem, len); + if (!data_find) { + data_find = prox_zmalloc(len, socket); + PROX_PANIC(data_find == NULL, "Failed to allocate memory (%u bytes) to hold header for peer\n", len); + + rte_memcpy(data_find, data_mem, len); + hash_set_add(hs, data_find, len); + } + + free(data_mem); + + *mem = data_find; + prox_sh_add_socket(socket, name, *mem); + return 0; +} + +static int lua_to_peer_data(struct lua_State *L, enum lua_place from, const char *name, uint32_t socket, struct peer_data *peer_data, size_t *cl, struct hash_set *hs) +{ + uint32_t hdr_len, hdr_beg, content_len, content_beg; + char hdr_file[256], content_file[256]; + int pop; + + if ((pop = lua_getfrom(L, from, name)) < 0) + return -1; + + if (!lua_istable(L, -1)) + return -1; + + if (lua_getfrom(L, TABLE, "header") < 0) + return -1; + if (lua_to_int(L, TABLE, "len", &hdr_len) < 0) + return -1; + if (lua_to_int(L, TABLE, "beg", &hdr_beg) < 0) + return -1; + if (lua_to_string(L, TABLE, "file_name", hdr_file, sizeof(hdr_file)) < 0) + return -1; + lua_pop(L, 1); + + if (lua_getfrom(L, TABLE, "content") < 0) + return -1; + if (lua_to_int(L, TABLE, "len", &content_len) < 0) + return -1; + if (lua_to_int(L, TABLE, "beg", &content_beg) < 0) + return -1; + if (lua_to_string(L, TABLE, "file_name", content_file, sizeof(content_file)) < 0) + return -1; + lua_pop(L, 1); + + if (hdr_len == UINT32_MAX) { + long ret = file_get_size(hdr_file); + + if (ret < 0) { + plog_err("%s", file_get_error()); + return -1; + } + hdr_len = ret - hdr_beg; + } + + if (content_len == UINT32_MAX) { + long ret = file_get_size(content_file); + + if (ret < 0) { + plog_err("%s", file_get_error()); + return -1; + } + content_len = ret - content_beg; + } + *cl = content_len; + peer_data->hdr_len = hdr_len; + + if (file_read_cached(hdr_file, &peer_data->hdr, hdr_beg, hdr_len, socket, hs)) + return -1; + if (file_read_cached(content_file, &peer_data->content, content_beg, content_len, socket, hs)) + return -1; + + lua_pop(L, pop); + return 0; +} + +static int lua_to_peer_action(struct lua_State *L, enum lua_place from, const char *name, struct peer_action *action, size_t client_contents_len, size_t server_contents_len) +{ + int pop; + + if ((pop = lua_getfrom(L, from, name)) < 0) + return -1; + + if (!lua_istable(L, -1)) + return -1; + + uint32_t peer, beg, len; + if (lua_to_int(L, TABLE, "peer", &peer) || + lua_to_int(L, TABLE, "beg", &beg) || + lua_to_int(L, TABLE, "len", &len)) { + return -1; + } + size_t data_len = (peer == PEER_CLIENT? client_contents_len : server_contents_len); + if (len == (uint32_t)-1) + len = data_len - beg; + + PROX_PANIC(beg + len > data_len, "Accessing data past the end (starting at %u for %u bytes) while total length is %zu\n", beg, len, data_len); + + action->peer = peer; + action->beg = beg; + action->len = len; + lua_pop(L, pop); + return 0; +} + +static int lua_to_stream_cfg(struct lua_State *L, enum lua_place from, const char *name, uint32_t socket, struct stream_cfg **stream_cfg, struct hash_set *hs) +{ + int pop; + struct stream_cfg *ret; + + if ((pop = lua_getfrom(L, from, name)) < 0) + return -1; + + if (lua_getfrom(L, TABLE, "actions") < 0) + return -1; + + lua_len(prox_lua(), -1); + uint32_t n_actions = lua_tointeger(prox_lua(), -1); + lua_pop(prox_lua(), 1); + + lua_pop(L, 1); + + size_t mem_size = 0; + mem_size += sizeof(*ret); + /* one additional action is allocated to allow inserting an + additional "default" action to close down TCP sessions from + the client side. */ + mem_size += sizeof(ret->actions[0]) * (n_actions + 1); + + ret = prox_zmalloc(sizeof(*ret) + mem_size, socket); + ret->n_actions = n_actions; + + size_t client_contents_len, server_contents_len; + char proto[16]; + uint32_t timeout_us, timeout_time_wait_us; + plogx_dbg("loading stream\n"); + if (lua_to_host_set(L, TABLE, "servers", &ret->servers)) + return -1; + if (lua_to_string(L, TABLE, "l4_proto", proto, sizeof(proto))) + return -1; + if (lua_to_peer_data(L, TABLE, "client_data", socket, &ret->data[PEER_CLIENT], &client_contents_len, hs)) + return -1; + if (lua_to_peer_data(L, TABLE, "server_data", socket, &ret->data[PEER_SERVER], &server_contents_len, hs)) + return -1; + + if (lua_to_int(L, TABLE, "timeout", &timeout_us)) { + timeout_us = 1000000; + } + + ret->tsc_timeout = usec_to_tsc(timeout_us); + + double up, dn; + + if (lua_to_double(L, TABLE, "up_bps", &up)) + up = 5000;// Default rate is 40 Mbps + + if (lua_to_double(L, TABLE, "dn_bps", &dn)) + dn = 5000;// Default rate is 40 Mbps + + const uint64_t hz = rte_get_tsc_hz(); + + ret->tt_cfg[PEER_CLIENT] = token_time_cfg_create(up, hz, ETHER_MAX_LEN + 20); + ret->tt_cfg[PEER_SERVER] = token_time_cfg_create(dn, hz, ETHER_MAX_LEN + 20); + + if (!strcmp(proto, "tcp")) { + ret->proto = IPPROTO_TCP; + ret->proc = stream_tcp_proc; + ret->is_ended = stream_tcp_is_ended; + + if (lua_to_int(L, TABLE, "timeout_time_wait", &timeout_time_wait_us)) { + timeout_time_wait_us = 2000000; + } + + ret->tsc_timeout_time_wait = usec_to_tsc(timeout_time_wait_us); + } + else if (!strcmp(proto, "udp")) { + plogx_dbg("loading UDP\n"); + ret->proto = IPPROTO_UDP; + ret->proc = stream_udp_proc; + ret->is_ended = stream_udp_is_ended; + } + else + return -1; + + /* get all actions */ + if (lua_getfrom(L, TABLE, "actions") < 0) + return -1; + + uint32_t idx = 0; + lua_pushnil(L); + while (lua_next(L, -2)) { + if (lua_to_peer_action(L, STACK, NULL, &ret->actions[idx], client_contents_len, server_contents_len)) + return -1; + + stream_cfg_verify_action(ret, &ret->actions[idx]); + + idx++; + + lua_pop(L, 1); + } + lua_pop(L, 1); + + /* For TCP, one of the peers initiates closing down the + connection. This is signified by the last action having + with zero length. If such an action is not specified in the + configuration file, the default is for the client to close + the connection. This means that the TCP connection at the + client will go into a TIME_WAIT state and the server + releases all the resources avoiding resource starvation at + the server. */ + if (ret->proto == IPPROTO_TCP && ret->actions[ret->n_actions - 1].len != 0) { + ret->actions[ret->n_actions].len = 0; + ret->actions[ret->n_actions].beg = 0; + ret->actions[ret->n_actions].peer = PEER_CLIENT; + ret->n_actions++; + } + + if (IPPROTO_TCP == ret->proto) + stream_tcp_calc_len(ret, &ret->n_pkts, &ret->n_bytes); + else + stream_udp_calc_len(ret, &ret->n_pkts, &ret->n_bytes); + + lua_pop(L, pop); + *stream_cfg = ret; + return 0; +} + +static int lua_to_bundle_cfg(struct lua_State *L, enum lua_place from, const char *name, uint8_t socket, struct bundle_cfg *bundle, struct hash_set *hs) +{ + int pop, pop2, idx; + int clients_loaded = 0; + + if ((pop = lua_getfrom(L, from, name)) < 0) + return -1; + + if (!lua_istable(L, -1)) + return -1; + + lua_len(prox_lua(), -1); + bundle->n_stream_cfgs = lua_tointeger(prox_lua(), -1); + lua_pop(prox_lua(), 1); + + bundle->stream_cfgs = prox_zmalloc(sizeof(*bundle->stream_cfgs) * bundle->n_stream_cfgs, socket); + + plogx_dbg("loading bundle cfg with %d streams\n", bundle->n_stream_cfgs); + idx = 0; + lua_pushnil(L); + while (lua_next(L, -2)) { + if (!clients_loaded) { + if (lua_to_host_set(L, TABLE, "clients", &bundle->clients)) { + return -1; + } + clients_loaded = 1; + } + if (lua_to_stream_cfg(L, STACK, NULL, socket, &bundle->stream_cfgs[idx], hs)) { + return -1; + } + + ++idx; + lua_pop(L, 1); + } + + lua_pop(L, pop); + return 0; +} + +static void init_task_gen(struct task_base *tbase, struct task_args *targ) +{ + struct task_gen_server *task = (struct task_gen_server *)tbase; + const int socket_id = rte_lcore_to_socket_id(targ->lconf->id); + + static char name[] = "server_mempool"; + name[0]++; + task->mempool = rte_mempool_create(name, + 4*1024 - 1, MBUF_SIZE, + targ->nb_cache_mbuf, + sizeof(struct rte_pktmbuf_pool_private), + rte_pktmbuf_pool_init, NULL, + rte_pktmbuf_init, 0, + socket_id, 0); + PROX_PANIC(task->mempool == NULL, "Failed to allocate memory pool with %u elements\n", 4*1024 - 1); + int pop = lua_getfrom(prox_lua(), GLOBAL, targ->streams); + PROX_PANIC(pop < 0, "Failed to find '%s' in lua\n", targ->streams); + + lua_len(prox_lua(), -1); + uint32_t n_listen = lua_tointeger(prox_lua(), -1); + lua_pop(prox_lua(), 1); + PROX_PANIC(n_listen == 0, "No services specified to listen on\n"); + + task->bundle_cfgs = prox_zmalloc(n_listen * sizeof(task->bundle_cfgs[0]), socket_id); + + plogx_info("n_listen = %d\n", n_listen); + + struct hash_set *hs = prox_sh_find_socket(socket_id, "genl4_streams"); + if (hs == NULL) { + /* Expected number of streams per bundle = 1, hash_set + will grow if full. */ + hs = hash_set_create(n_listen, socket_id); + prox_sh_add_socket(socket_id, "genl4_streams", hs); + } + + const struct rte_hash_parameters listen_table = { + .name = name, + .entries = n_listen * 4, + .key_len = sizeof(struct new_tuple), + .hash_func = rte_hash_crc, + .hash_func_init_val = 0, + .socket_id = socket_id, + }; + name[0]++; + + task->listen_hash = rte_hash_create(&listen_table); + task->listen_entries = prox_zmalloc(listen_table.entries * sizeof(task->listen_entries[0]), socket_id); + + int idx = 0; + lua_pushnil(prox_lua()); + while (lua_next(prox_lua(), -2)) { + task->bundle_cfgs[idx].n_stream_cfgs = 1; + task->bundle_cfgs[idx].stream_cfgs = prox_zmalloc(sizeof(*task->bundle_cfgs[idx].stream_cfgs), socket_id); + int ret = lua_to_stream_cfg(prox_lua(), STACK, NULL, socket_id, &task->bundle_cfgs[idx].stream_cfgs[0], hs); + PROX_PANIC(ret, "Failed to load stream cfg\n"); + struct stream_cfg *stream = task->bundle_cfgs[idx].stream_cfgs[0]; + + // TODO: check mask and add to hash for each host + struct new_tuple nt = { + .dst_addr = stream->servers.ip, + .proto_id = stream->proto, + .dst_port = stream->servers.port, + .l2_types[0] = 0x0008, + }; + + ret = rte_hash_add_key(task->listen_hash, &nt); + PROX_PANIC(ret < 0, "Failed to add\n"); + + task->listen_entries[ret] = &task->bundle_cfgs[idx]; + + plogx_dbg("Server = "IPv4_BYTES_FMT":%d\n", IPv4_BYTES(((uint8_t*)&nt.dst_addr)), rte_bswap16(nt.dst_port)); + ++idx; + lua_pop(prox_lua(), 1); + } + + static char name2[] = "task_gen_hash2"; + + name2[0]++; + plogx_dbg("Creating bundle ctx pool\n"); + if (bundle_ctx_pool_create(name2, targ->n_concur_conn * 2, &task->bundle_ctx_pool, NULL, 0, NULL, socket_id)) { + cmd_mem_stats(); + PROX_PANIC(1, "Failed to create conn_ctx_pool\n"); + } + + task->heap = heap_create(targ->n_concur_conn * 2, socket_id); + task->seed = rte_rdtsc(); + + /* TODO: calculate the CDF of the reply distribution and the + number of replies as the number to cover for 99% of the + replies. For now, assume that this is number is 2. */ + uint32_t queue_size = rte_align32pow2(targ->n_concur_conn * 2); + + PROX_PANIC(queue_size == 0, "Overflow resulted in queue size 0\n"); + task->fqueue = fqueue_create(queue_size, socket_id); + PROX_PANIC(task->fqueue == NULL, "Failed to allocate local queue\n"); + + uint32_t n_descriptors; + + if (targ->nb_txports) { + PROX_PANIC(targ->nb_txports != 1, "Need exactly one TX port for L4 generation\n"); + n_descriptors = prox_port_cfg[targ->tx_port_queue[0].port].n_txd; + } else { + PROX_PANIC(targ->nb_txrings != 1, "Need exactly one TX ring for L4 generation\n"); + n_descriptors = 256; + } + + struct token_time_cfg tt_cfg = { + .bpp = targ->rate_bps, + .period = rte_get_tsc_hz(), + .bytes_max = n_descriptors * (ETHER_MIN_LEN + 20), + }; + + token_time_init(&task->token_time, &tt_cfg); +} + +static void init_task_gen_client(struct task_base *tbase, struct task_args *targ) +{ + struct task_gen_client *task = (struct task_gen_client *)tbase; + static char name[] = "gen_pool"; + const uint32_t socket = rte_lcore_to_socket_id(targ->lconf->id); + name[0]++; + task->mempool = rte_mempool_create(name, + 4*1024 - 1, MBUF_SIZE, + targ->nb_cache_mbuf, + sizeof(struct rte_pktmbuf_pool_private), + rte_pktmbuf_pool_init, NULL, + rte_pktmbuf_init, 0, + socket, 0); + PROX_PANIC(task->mempool == NULL, "Failed to allocate memory pool with %u elements\n", 4*1024 - 1); + + /* streams contains a lua table. Go through it and read each + stream with associated imix_fraction. */ + uint32_t imix; + uint32_t i = 0; + + int pop = lua_getfrom(prox_lua(), GLOBAL, targ->streams); + PROX_PANIC(pop < 0, "Failed to find '%s' in lua\n", targ->streams); + + lua_len(prox_lua(), -1); + uint32_t n_bundle_cfgs = lua_tointeger(prox_lua(), -1); + lua_pop(prox_lua(), 1); + PROX_PANIC(n_bundle_cfgs == 0, "No configs specified\n"); + plogx_info("loading %d bundle_cfgs\n", n_bundle_cfgs); + + struct hash_set *hs = prox_sh_find_socket(socket, "genl4_streams"); + if (hs == NULL) { + /* Expected number of streams per bundle = 8, hash_set + will grow if full. */ + hs = hash_set_create(n_bundle_cfgs * 8, socket); + prox_sh_add_socket(socket, "genl4_streams", hs); + } + + task->bundle_cfgs = prox_zmalloc(n_bundle_cfgs * sizeof(task->bundle_cfgs[0]), socket); + lua_pushnil(prox_lua()); + + int total_imix = 0; + + uint32_t *occur = prox_zmalloc(n_bundle_cfgs * sizeof(*occur), socket); + struct cdf *cdf = cdf_create(n_bundle_cfgs, socket); + + while (lua_next(prox_lua(), -2)) { + PROX_PANIC(lua_to_int(prox_lua(), TABLE, "imix_fraction", &imix) || + lua_to_bundle_cfg(prox_lua(), TABLE, "bundle", socket, &task->bundle_cfgs[i], hs), + "Failed to load bundle cfg:\n%s\n", get_lua_to_errors()); + cdf_add(cdf, imix); + occur[i] = imix; + total_imix += imix; + ++i; + lua_pop(prox_lua(), 1); + } + + lua_pop(prox_lua(), pop); + cdf_setup(cdf); + + PROX_PANIC(targ->max_setup_rate == 0, "Max setup rate not set\n"); + + task->new_conn_cost = rte_get_tsc_hz()/targ->max_setup_rate; + + static char name2[] = "task_gen_hash"; + name2[0]++; + plogx_dbg("Creating bundle ctx pool\n"); + if (bundle_ctx_pool_create(name2, targ->n_concur_conn, &task->bundle_ctx_pool, occur, n_bundle_cfgs, task->bundle_cfgs, socket)) { + cmd_mem_stats(); + PROX_PANIC(1, "Failed to create conn_ctx_pool\n"); + } + + task->heap = heap_create(targ->n_concur_conn, socket); + task->seed = rte_rdtsc(); + /* task->token_time.bytes_max = MAX_PKT_BURST * (ETHER_MAX_LEN + 20); */ + + /* To avoid overflowing the tx descriptors, the token bucket + size needs to be limited. The descriptors are filled most + quickly with the smallest packets. For that reason, the + token bucket size is given by "number of tx descriptors" * + "smallest Ethernet packet". */ + PROX_ASSERT(targ->nb_txports == 1); + + struct token_time_cfg tt_cfg = { + .bpp = targ->rate_bps, + .period = rte_get_tsc_hz(), + .bytes_max = prox_port_cfg[targ->tx_port_queue[0].port].n_txd * (ETHER_MIN_LEN + 20), + }; + + token_time_init(&task->token_time, &tt_cfg); +} + +static void start_task_gen_client(struct task_base *tbase) +{ + struct task_gen_client *task = (struct task_gen_client *)tbase; + + token_time_reset(&task->token_time, rte_rdtsc(), 0); + + task->new_conn_tokens = 0; + task->new_conn_last_tsc = rte_rdtsc(); +} + +static void stop_task_gen_client(struct task_base *tbase) +{ + struct task_gen_client *task = (struct task_gen_client *)tbase; + struct bundle_ctx *bundle; + + while (!heap_is_empty(task->heap)) { + bundle = BUNDLE_CTX_UPCAST(heap_pop(task->heap)); + bundle_expire(bundle, &task->bundle_ctx_pool, &task->l4_stats); + } +} + +static void start_task_gen_server(struct task_base *tbase) +{ + struct task_gen_server *task = (struct task_gen_server *)tbase; + + token_time_reset(&task->token_time, rte_rdtsc(), 0); +} + +static void stop_task_gen_server(struct task_base *tbase) +{ + struct task_gen_server *task = (struct task_gen_server *)tbase; + struct bundle_ctx *bundle; + uint8_t out[MAX_PKT_BURST]; + + while (!heap_is_empty(task->heap)) { + bundle = BUNDLE_CTX_UPCAST(heap_pop(task->heap)); + bundle_expire(bundle, &task->bundle_ctx_pool, &task->l4_stats); + } + + if (task->cancelled) { + struct rte_mbuf *mbuf = task->mbuf_saved; + + out[0] = OUT_DISCARD; + task->cancelled = 0; + task->base.tx_pkt(&task->base, &mbuf, 1, out); + } + + do { + if (task->cur_mbufs_beg == task->cur_mbufs_end) { + task->cur_mbufs_end = fqueue_get(task->fqueue, task->cur_mbufs, MAX_PKT_BURST); + task->cur_mbufs_beg = 0; + if (task->cur_mbufs_end == 0) + break; + } + uint16_t n_pkts = task->cur_mbufs_end - task->cur_mbufs_beg; + struct rte_mbuf **mbufs = task->cur_mbufs + task->cur_mbufs_beg; + + if (n_pkts) { + for (uint16_t j = 0; j < n_pkts; ++j) { + out[j] = OUT_DISCARD; + } + task->base.tx_pkt(&task->base, mbufs, n_pkts, out); + } + } while (1); +} + +static struct task_init task_init_gen1 = { + .mode_str = "genl4", + .sub_mode_str = "server", + .init = init_task_gen, + .handle = handle_gen_bulk, + .start = start_task_gen_server, + .stop = stop_task_gen_server, + .flag_features = TASK_FEATURE_ZERO_RX, + .size = sizeof(struct task_gen_server), + .mbuf_size = 2048 + sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM, +}; + +static struct task_init task_init_gen2 = { + .mode_str = "genl4", + .init = init_task_gen_client, + .handle = handle_gen_bulk_client, + .start = start_task_gen_client, + .stop = stop_task_gen_client, + .flag_features = TASK_FEATURE_ZERO_RX, + .size = sizeof(struct task_gen_client), + .mbuf_size = 2048 + sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM, +}; + +__attribute__((constructor)) static void reg_task_gen(void) +{ + reg_task(&task_init_gen1); + reg_task(&task_init_gen2); +} |