#include #include #include #include #include #include "log.h" #include "marsio.h" #include "marsio_io.h" #include "packet_def.h" #include "packet_utils.h" #include "packet_parse.h" #define PACKET_IO_LOG_ERROR(format, ...) LOG_ERROR("marsio", format, ##__VA_ARGS__) struct marsio_io { struct mr_instance *mr_ins; struct mr_vdev *mr_dev; struct mr_sendpath *mr_path; struct packet_io_stat stat[MAX_THREAD_NUM]; }; /****************************************************************************** * Private API ******************************************************************************/ static void metadata_from_mbuff_to_packet(marsio_buff_t *mbuff, struct packet *pkt) { struct route_ctx route_ctx = {0}; struct sids sids = {0}; uint64_t session_id = {0}; uint64_t domain = {0}; uint16_t link_id = {0}; int is_ctrl = {0}; enum packet_direction direction = PACKET_DIRECTION_OUTGOING; route_ctx.used = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, &route_ctx.data, sizeof(route_ctx.data)); if (route_ctx.used > 0) { packet_set_route_ctx(pkt, &route_ctx); } else { PACKET_IO_LOG_ERROR("failed to get route ctx"); } sids.used = marsio_buff_get_sid_list(mbuff, sids.sid, sizeof(sids.sid) / sizeof(sids.sid[0])); if (sids.used > 0) { packet_set_sids(pkt, &sids); } else { PACKET_IO_LOG_ERROR("failed to get sids"); } if (marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) == sizeof(session_id)) { packet_set_session_id(pkt, session_id); } else { PACKET_IO_LOG_ERROR("failed to get session id"); } // TODO #if 0 if (marsio_buff_get_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) == sizeof(domain)) { packet_set_domain(pkt, domain); } else { PACKET_IO_LOG_ERROR("failed to get domain id"); } #endif if (marsio_buff_get_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) == sizeof(link_id)) { packet_set_link_id(pkt, link_id); } else { PACKET_IO_LOG_ERROR("failed to get link id"); } is_ctrl = marsio_buff_is_ctrlbuf(mbuff); packet_set_ctrl(pkt, is_ctrl); if (marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) == sizeof(direction)) { packet_set_direction(pkt, direction); } else { PACKET_IO_LOG_ERROR("failed to get direction"); } packet_set_action(pkt, PACKET_ACTION_FORWARD); packet_set_origin_ctx(pkt, mbuff); } static void metadata_from_packet_to_mbuff(struct packet *pkt, marsio_buff_t *mbuff) { const struct route_ctx *route_ctx = packet_get_route_ctx(pkt); const struct sids *sids = packet_get_sids(pkt); uint64_t session_id = packet_get_session_id(pkt); uint64_t domain = packet_get_domain(pkt); uint16_t link_id = packet_get_link_id(pkt); int is_ctrl = packet_is_ctrl(pkt); enum packet_direction direction = packet_get_direction(pkt); if (marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, (void *)route_ctx->data, route_ctx->used) != 0) { PACKET_IO_LOG_ERROR("failed to set route ctx"); } if (marsio_buff_set_sid_list(mbuff, (sid_t *)sids->sid, sids->used) != 0) { PACKET_IO_LOG_ERROR("failed to set sids"); } if (marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) != 0) { PACKET_IO_LOG_ERROR("failed to set session id"); } // TODO #if 0 if (marsio_buff_set_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) != 0) { PACKET_IO_LOG_ERROR("failed to set domain id"); } #endif if (marsio_buff_set_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) != 0) { PACKET_IO_LOG_ERROR("failed to set link id"); } if (is_ctrl) { marsio_buff_set_ctrlbuf(mbuff); } if (marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) != 0) { PACKET_IO_LOG_ERROR("failed to set direction"); } } static inline int is_keepalive_packet(const char *data, int len) { if (data == NULL || len < (int)(sizeof(struct ethhdr))) { return 0; } struct ethhdr *eth_hdr = (struct ethhdr *)data; if (eth_hdr->h_proto == 0xAAAA) { return 1; } else { return 0; } } /****************************************************************************** * Public API ******************************************************************************/ struct marsio_io *marsio_io_new(const char *app_symbol, const char *dev_symbol, uint16_t *cpu_mask, uint16_t nr_threads) { int opt = 1; cpu_set_t coremask; CPU_ZERO(&coremask); for (uint16_t i = 0; i < nr_threads; i++) { CPU_SET(cpu_mask[i], &coremask); } struct marsio_io *handle = (struct marsio_io *)calloc(1, sizeof(struct marsio_io)); if (handle == NULL) { PACKET_IO_LOG_ERROR("unable to allocate memory for marsio_io"); return NULL; } handle->mr_ins = marsio_create(); if (handle->mr_ins == NULL) { PACKET_IO_LOG_ERROR("unable to create marsio instance"); goto error_out; } marsio_option_set(handle->mr_ins, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &coremask, sizeof(cpu_set_t)); marsio_option_set(handle->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt)); if (marsio_init(handle->mr_ins, app_symbol) != 0) { PACKET_IO_LOG_ERROR("unable to init marsio instance"); goto error_out; } handle->mr_dev = marsio_open_device(handle->mr_ins, dev_symbol, nr_threads, nr_threads); if (handle->mr_dev == NULL) { PACKET_IO_LOG_ERROR("unable to open marsio device"); goto error_out; } handle->mr_path = marsio_sendpath_create_by_vdev(handle->mr_dev); if (handle->mr_path == NULL) { PACKET_IO_LOG_ERROR("unable to create marsio sendpath"); goto error_out; } return handle; error_out: marsio_io_free(handle); return NULL; } void marsio_io_free(struct marsio_io *handle) { if (handle) { if (handle->mr_path) { marsio_sendpath_destory(handle->mr_path); handle->mr_path = NULL; } if (handle->mr_dev) { marsio_close_device(handle->mr_dev); handle->mr_dev = NULL; } if (handle->mr_ins) { marsio_destory(handle->mr_ins); handle->mr_ins = NULL; } free(handle); handle = NULL; } } int marsio_io_init(struct marsio_io *handle, uint16_t thr_idx) { if (marsio_thread_init(handle->mr_ins) != 0) { PACKET_IO_LOG_ERROR("unable to init marsio thread"); return -1; } return 0; } int marsio_io_ingress(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { struct packet *pkt; marsio_buff_t *mbuff; marsio_buff_t *rx_buffs[RX_BURST_MAX]; struct packet_io_stat *stat = &handle->stat[thr_idx]; int nr_recv; int nr_parsed = 0; int len; char *data; nr_recv = marsio_recv_burst(handle->mr_dev, thr_idx, rx_buffs, MIN(RX_BURST_MAX, nr_pkts)); if (nr_recv <= 0) { return 0; } for (int i = 0; i < nr_recv; i++) { mbuff = rx_buffs[i]; data = marsio_buff_mtod(mbuff); len = marsio_buff_datalen(mbuff); stat->dev_rx_pkts++; stat->dev_rx_bytes += len; if (is_keepalive_packet(data, len)) { stat->keep_alive_pkts++; stat->keep_alive_bytes += len; stat->dev_tx_pkts++; stat->dev_tx_bytes += len; marsio_send_burst(handle->mr_path, thr_idx, &mbuff, 1); continue; } pkt = &pkts[nr_parsed]; //memset(pkt, 0, sizeof(struct packet)); packet_parse(pkt, data, len); metadata_from_mbuff_to_packet(mbuff, pkt); nr_parsed++; if (marsio_buff_is_ctrlbuf(mbuff)) { stat->ctrl_rx_pkts++; stat->ctrl_rx_bytes += len; } else { stat->raw_rx_pkts++; stat->raw_rx_bytes += len; } } return nr_parsed; } void marsio_io_egress(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { struct packet *pkt; marsio_buff_t *mbuff; struct packet_io_stat *stat = &handle->stat[thr_idx]; int len; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; len = packet_get_raw_len(pkt); stat->dev_tx_pkts++; stat->dev_tx_bytes += len; mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt); assert(mbuff != NULL); metadata_from_packet_to_mbuff(pkt, mbuff); if (marsio_buff_is_ctrlbuf(mbuff)) { stat->ctrl_tx_pkts++; stat->ctrl_tx_bytes += len; } else { stat->raw_tx_pkts++; stat->raw_tx_bytes += len; } marsio_send_burst(handle->mr_path, thr_idx, &mbuff, 1); packet_free(pkt); } } void marsio_io_drop(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { struct packet *pkt; marsio_buff_t *mbuff; struct packet_io_stat *stat = &handle->stat[thr_idx]; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt); if (mbuff) { stat->drop_pkts++; stat->drop_bytes += packet_get_raw_len(pkt); marsio_buff_free(handle->mr_ins, &mbuff, 1, 0, thr_idx); } packet_free(pkt); } } int marsio_io_inject(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { int len; int nr_inject = 0; char *ptr; struct packet *pkt; marsio_buff_t *mbuff; struct packet_io_stat *stat = &handle->stat[thr_idx]; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; len = packet_get_raw_len(pkt); if (marsio_buff_malloc_global(handle->mr_ins, &mbuff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0) { PACKET_IO_LOG_ERROR("unable to allocate marsio buffer for inject packet"); continue; } stat->inject_pkts++; stat->inject_bytes += len; stat->raw_tx_pkts++; stat->raw_tx_bytes += len; stat->dev_tx_pkts++; stat->dev_tx_bytes += len; nr_inject++; ptr = marsio_buff_append(mbuff, len); memcpy(ptr, packet_get_raw_data(pkt), len); metadata_from_packet_to_mbuff(pkt, mbuff); marsio_send_burst_with_options(handle->mr_path, thr_idx, &mbuff, 1, MARSIO_SEND_OPT_REHASH); packet_free(pkt); } return nr_inject; } void marsio_io_yield(struct marsio_io *handle, uint16_t thr_idx, uint64_t timeout_ms) { struct mr_vdev *vdevs[1] = { handle->mr_dev, }; marsio_poll_wait(handle->mr_ins, vdevs, 1, thr_idx, timeout_ms); } struct packet_io_stat *marsio_io_stat(struct marsio_io *handle, uint16_t thr_idx) { return &handle->stat[thr_idx]; }