path: root/pag/libpag.c

/*
 *	Packet Acquisition and Generation Library for CNCERT
 *	Author : Qiuwen Lu<[email protected]>
 *	Date   : 2016-09-12
 */

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <assert.h>
#include <netinet/in.h>
#include <MESA_prof_load.h>

#include <rte_ether.h>
#include <mr_common.h>
#include <marsio.h>

#include "libpag.h"

#define PAG_SYMBOL_MAX		64
#define PAG_STRING_MAX		1024
#define PAG_BURST_MAX		64
#define PAG_DEVICE_MAX		16
#define PAG_CPU_ID_MAX		64

struct pag_instance
{
	char app_name[PAG_SYMBOL_MAX];
	char dev_name[PAG_STRING_MAX];
	unsigned int nr_devices;
	unsigned int nr_rx_stream;
	unsigned int nr_tx_stream;
	unsigned int burst_rx;
	unsigned int burst_tx;
	unsigned int autoexit;
	unsigned int looptimes;
	uint64_t coremask;
};

struct pag_thread_instance
{
	int thread_inited;
	marsio_buff_t * rxmbuf[PAG_BURST_MAX];
	marsio_buff_t * txmbuf[PAG_BURST_MAX];
	unsigned int rxmbuf_max;
	unsigned int txmbuf_max;
	unsigned int rxmbuf_used;
	unsigned int txmbuf_used;
	unsigned int rxmbuf_cur;
	unsigned int txmbuf_cur;
	unsigned int cur_device;

	// �����Ļ����
	marsio_buff_t * rxmbuf_ctx;
};

static int __pag_inited = 0;
static __thread struct pag_thread_instance thread_instance_;

struct pag_instance pag_config_;

#define PAG_LOG(fmt, ...)									\
do {														\
	fprintf(stderr, "libpag: " fmt "\n",## __VA_ARGS__);	\
} while(0)													\

#define PAG_CFGFILE					"pag.conf"
#define PAG_DEFAULT_RX_BURST		32
#define PAG_DEFAULT_TX_BURST		32
#define PAG_DEFAULT_AUTOEXIT		0
#define PAG_DEFAULT_LOOP			0

int __strsplit(char *string, int stringlen, char **tokens, int maxtokens, char delim)
{
	int i, tok = 0;
	int tokstart = 1; /* first token is right at start of string */

	if (string == NULL || tokens == NULL)
		goto einval_error;

	for (i = 0; i < stringlen; i++) {
		if (string[i] == '\0' || tok >= maxtokens)
			break;
		if (tokstart) {
			tokstart = 0;
			tokens[tok++] = &string[i];
		}
		if (string[i] == delim) {
			string[i] = '\0';
			tokstart = 1;
		}
	}
	return tok;

einval_error:
	errno = EINVAL;
	return -1;
}

static int pag_config_load_app_info(struct pag_instance * instance)
{
	int ret = MESA_load_profile_string_nodef(PAG_CFGFILE, "pag", "app_name",
		instance->app_name, sizeof(instance->app_name));

	if (ret < 0)
	{
		PAG_LOG("load pag.conf failed(section=pag, key=app_name).");
		return -1;
	}

	return 0;
}

static int pag_config_load_stream_info(struct pag_instance * instance)
{
	unsigned int cpu_id_range[PAG_CPU_ID_MAX] = { 0 };
	
	// ��CPU�����
	int cpu_id_count = MESA_load_profile_uint_range(PAG_CFGFILE, "pag", "cpu_id",
		PAG_CPU_ID_MAX, cpu_id_range);
	
	if (cpu_id_count < 0)
	{
		PAG_LOG("cpu_id is missing, please recheck %s", PAG_CFGFILE);
		return -1;
	}

	uint64_t coremask = 0;
	for(int i = 0; i < cpu_id_count; i++)
	{
		coremask |= 1ULL << cpu_id_range[i];
	}
	
	// ��Ĭ���豸������û��ָ��ʹ���߳���
	unsigned int nr_rx_stream;
	unsigned int nr_tx_stream;
	unsigned int nr_stream_default = cpu_id_count;
	
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "rxstream",
		&nr_rx_stream, nr_stream_default);
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "txstream",
		&nr_tx_stream, nr_stream_default);
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "autoexit",
		&instance->autoexit, PAG_DEFAULT_AUTOEXIT);

	instance->coremask = coremask;
	instance->nr_rx_stream = nr_rx_stream;
	instance->nr_tx_stream = nr_tx_stream;

	marsio_option_set(instance->app_name, MARSIO_OPT_THREAD_MASK,
		&instance->coremask, sizeof(instance->coremask));
	marsio_option_set(instance->app_name, MARSIO_OPT_AUTOEXIT,
		&instance->autoexit, sizeof(instance->autoexit));
	marsio_option_set(instance->app_name, MARSIO_OPT_RAW_RX_STREAM_NUM,
		&instance->nr_rx_stream, sizeof(instance->nr_rx_stream));
	marsio_option_set(instance->app_name, MARSIO_OPT_RAW_TX_STREAM_NUM,
		&instance->nr_tx_stream, sizeof(instance->nr_tx_stream));

	PAG_LOG("coremask=%"PRIx64", rxstream=%u, txstream=%u, autoexit=%d",
		instance->coremask, instance->nr_rx_stream, instance->nr_tx_stream,
		instance->autoexit);

	return 0;
}

static int pag_config_load_burst_info(struct pag_instance * instance)
{
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "burst_rx", &instance->burst_rx, 
		PAG_DEFAULT_RX_BURST);
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "burst_tx", &instance->burst_tx, 
		PAG_DEFAULT_TX_BURST);
	MESA_load_profile_uint_def(PAG_CFGFILE, "pag", "loop", &instance->looptimes,
		PAG_DEFAULT_LOOP);

	if(instance->burst_rx > PAG_BURST_MAX)
	{
		PAG_LOG("burst_rx=%d is larger than limit(limit=%d), please recheck %s",
			instance->burst_rx, PAG_BURST_MAX, PAG_CFGFILE);
		return -1;
	}

	if(instance->burst_tx > PAG_BURST_MAX)
	{
		PAG_LOG("burst_tx=%d is larger than limit(limit=%d), please recheck %s",
			instance->burst_tx, PAG_BURST_MAX, PAG_CFGFILE);
		return -2;
	}
	
	PAG_LOG("burst_rx=%d, burst_tx=%d", instance->burst_rx, instance->burst_tx);
	return 0;
}

static int pag_config_load_device_info(struct pag_instance * instance)
{
	int ret = MESA_load_profile_string_nodef(PAG_CFGFILE, "pag", "dev_name",
		instance->dev_name, sizeof(instance->dev_name));

	if (ret < 0)
	{
		PAG_LOG("load pag.conf failed(section = pag, key = dev_name).");
		return -1;
	}

	char * str_devices[PAG_DEVICE_MAX];
	int nr_devices;

	nr_devices = __strsplit(instance->dev_name, sizeof(instance->dev_name),
		str_devices, PAG_DEVICE_MAX, ',');

	if (nr_devices <= 0)
	{
		PAG_LOG("load pag.conf failed(section = pag, key = dev_name), "
			"invailed format");
		return -2;
	}

	for (int i = 0; i < nr_devices; i++)
		marsio_option_set(instance->app_name, MARSIO_OPT_RAW_DEVICE, 
			str_devices[i], sizeof(str_devices[i]));

	instance->nr_devices = nr_devices;
	return 0;
}

static int pag_thread_init(struct pag_instance * instance,
	struct pag_thread_instance * tinstance)
{
	if (likely(tinstance->thread_inited != 0))
		return 0;

	marsio_thread_init();
	tinstance->rxmbuf_max = instance->burst_rx;
	tinstance->txmbuf_max = instance->burst_tx;
	tinstance->thread_inited = 1;
	return 0;
}

static int pag_config(struct pag_instance * instance)
{
	if (access(PAG_CFGFILE, R_OK) != 0)
	{
		PAG_LOG("load pag.conf failed, pag.conf doesnot existed.");
		return -1;
	}
	
	int ret = pag_config_load_app_info(instance);
	if (ret < 0) return ret;
	ret = pag_config_load_burst_info(instance);
	if (ret < 0) return ret;
	ret = pag_config_load_stream_info(instance);
	if (ret < 0) return ret;
	ret = pag_config_load_device_info(instance);
	if (ret < 0) return ret;
	return 0;
}

int pag_open()
{
	if (__pag_inited != 0) return 0;

	struct pag_instance * instance = &pag_config_;
	
	int ret = pag_config(instance);
	if(ret < 0)
	{
		PAG_LOG("load configure failed.");
		return -1;
	}

	ret = marsio_init(instance->app_name);
	
	if(ret < 0)
	{
		PAG_LOG("marsio library init failed(ret=%d).", ret);
		return -3;
	}

	if (ret < 0)
		exit(EXIT_FAILURE);
	
	__pag_inited = 1;
	return 0;
}

static void inline __free_frames(struct pag_thread_instance * tinstance)
{
	marsio_buff_free(tinstance->rxmbuf, tinstance->rxmbuf_used);
	return;
}

// �������豸��ȡ���ݰ����ڱ��̻߳�����Ϊ��ʱ����
static int inline __get_frames_from_rtdevice(struct pag_instance * instance,
	struct pag_thread_instance * tinstance, int sid)
{
	// �α���ڵ����ϴ��ձ�����ʱ��˵����һ���ձ��ı���ȫ������
	assert((tinstance->rxmbuf_cur >= tinstance->rxmbuf_used));

	// һ����rxmbuf_max����������һ��������
	unsigned int cur_device = tinstance->cur_device;
	int ret = marsio_raw_recv_burst(cur_device, sid, tinstance->rxmbuf, 
		tinstance->rxmbuf_max);

	if (ret < 0) return ret;

	// ���û��������л��ձ��豸
	tinstance->cur_device = (cur_device + 1) % instance->nr_devices;
	tinstance->rxmbuf_cur = 0;
	tinstance->rxmbuf_used = ret;
	
	assert(tinstance->rxmbuf_used <= tinstance->rxmbuf_max);
	return ret;
}

#define __PAG_LOOP_TIMES		5

void * pag_get_frame(int sid)
{
	struct pag_thread_instance * tinstance = &thread_instance_;
	struct pag_instance * instance = &pag_config_;
	pag_thread_init(instance, tinstance);
	
	// �����������ݰ������꣬���ͷţ�Ȼ���ٴ�����ȡ
	if (unlikely((tinstance->rxmbuf_cur >= tinstance->rxmbuf_used)))
	{
		__free_frames(tinstance);
		__get_frames_from_rtdevice(instance, tinstance, sid);
	}
	
	int loop_times = 0;
	while (tinstance->rxmbuf_used == 0 && loop_times < instance->looptimes)
	{
		__get_frames_from_rtdevice(instance, tinstance, sid);
		loop_times++;
	}

	if (unlikely(tinstance->rxmbuf_used == 0)) return NULL;
	tinstance->rxmbuf_ctx = tinstance->rxmbuf[tinstance->rxmbuf_cur++];
	return (void *)marsio_buff_mtod(tinstance->rxmbuf_ctx);
}

int pag_get_frame_length(int sid)
{
	struct pag_thread_instance * tinstance = &thread_instance_;
	if (unlikely(tinstance->rxmbuf_ctx == NULL)) return 0;
	return marsio_buff_datalen(tinstance->rxmbuf_ctx);
}

void * pag_get_frame_with_len(int sid, unsigned int * len)
{
	void * frame = pag_get_frame(sid);
	*len = pag_get_frame_length(sid);
	return frame;
}

void * pag_get(int sid)
{
	void * frame = pag_get_frame(sid);
	if (frame == NULL) return NULL;

	struct ether_hdr * eth_hdr = (struct ether_hdr *)frame;
	switch(ntohs(eth_hdr->ether_type))
	{
	case ETHER_TYPE_IPv4:
	case ETHER_TYPE_IPv6:
		return (void *)((uint8_t *)frame + sizeof(struct ether_hdr));
	default:
		return NULL;
	}
}

int pag_close()
{
	marsio_destory();
	return 0;
}

void * pag_getsendbuf(int sid)
{
	return NULL;
}

int pag_send(int pkttype, int sid, int datalen)
{
	return 0;
}

void pag_freesendbuf(int sid)
{
}

void * pag_getsendbuf_eth(int sid, int port)
{
	return NULL;
}

int pag_send_eth(int sid, int eth_datalen, int port)
{
	return 0;
}

void pag_freesendbuf_eth(int sid, int port)
{
}

uint64_t pag_time(void * pkt)
{
	return 0;
}