path: root/src/tcpreplay.c

/* $Id: tcpreplay.c 2433 2010-03-28 20:57:36Z aturner $ */

/*
 * Copyright (c) 2001-2010 Aaron Turner.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. Neither the names of the copyright owners nor the names of its
 *    contributors may be used to endorse or promote products derived from
 *    this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE
 * GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
 * IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/*
 * 2013-06-06, LiJia modify for replay packet multiply.
 * add two new arguments:
 *  -m,  Replay streams at a given multiple;
 *  -d,  The distance between raw packet and fake replay packet.
 *  -D,  assign tcpreplay send packet driver, support pcap or marsio.
 *  -s,  marsio CPU bind mask.
 */ 

#include "config.h"
#include "defines.h"
#include "common.h"

#include <ctype.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <unistd.h>
#include <errno.h>

#ifndef __USE_BSD
#define __USE_BSD 1
#endif
#include <netinet/ip.h>
#ifndef __FAVOR_BSD
#define __FAVOR_BSD 1
#endif
#include <netinet/tcp.h>
#include <netinet/udp.h>

#include "tcpreplay.h"

#ifdef TCPREPLAY_EDIT
#include "tcpreplay_edit_opts.h"
#include "tcpedit/tcpedit.h"
tcpedit_t *tcpedit;
#else
#include "tcpreplay_opts.h"
#endif

#include "send_packets.h"
#include "signal_handler.h"


tcpreplay_opt_t options;
struct timeval begin, end;
COUNTER bytes_sent, failed, pkts_sent;
int cache_bit, cache_byte;
volatile int didsig;

#ifdef DEBUG
int debug = 0;
#endif

#ifdef HAVE_ABSOLUTE_TIME
#include <CoreServices/CoreServices.h>
#endif

void preload_pcap_file(int file_idx);
void replay_file(int file_idx);
void usage(void);
void init(void);
void post_args(void);

extern int send_pkt_driver_mode; /* 1:pcap; 12:marsio4; */

#ifdef TCPBURST

int tcpburst_version_VERSION_20180611;

#define PROCESS_BAR_SW	(1) /* 实时进度条 */
#define TCP_BURST_MTU		(1514)

#include <linux/types.h>
#include <linux/stddef.h>
#include <assert.h>
#include "MESA_list.h"

#ifdef __cplusplus
extern "C"
{
#endif

#if PROCESS_BAR_SW /* 2013-12-04 LiJia add, for 读大包进度实时显示 */
static char *dumpfile_name;
static int dumpfile_last_file_index = -1;
static unsigned long dumpfile_total_size;
static unsigned long dumpfile_read_size = 0;
static int dump_times = 0;
static char process_bar[100][101];

static unsigned long __get_file_length(const char *file_name)
{
	struct stat file_stat;
	
	if(stat(file_name, &file_stat) == 0)
	{
		return file_stat.st_size;
	}
	return 0;
}

void process_bar_print(int len, const char *filename)
{
	double percent;
	int index;

	if(dumpfile_total_size < 512 * 10000){
		return;
	}
	
	dump_times++;
	dumpfile_read_size += len + sizeof(struct pcap_pkthdr);

	if((dump_times % 10240) == 0){
		if(dumpfile_read_size < dumpfile_total_size){
			percent = ((double)dumpfile_read_size/(double)dumpfile_total_size) * 100;
		}else{
			percent = 100.00;
		}
		index = (int )percent;
		if(index >= 99){
			index = 99;
		}
		printf("\033[HReply dumpfile '%s':\n", filename);
		printf("%s \033[41m%0.2f%% \033[0m\n", process_bar[index], percent);
	}
}

void process_bar_prt_all(void)
{
	int i;
	for(i = 0; i < 100; i++){
		printf("%s\n", process_bar[i]);
	}
}

int process_bar_prt_init(void)
{
	int i, j;

	for(i = 0; i < 100; i++){
		for(j = 0; j < 100; j++){
			if(j <= i){
				process_bar[i][j] = '>';
			}else{
				process_bar[i][j] = '.';
			}
		}
		process_bar[i][100] = '\0'; 
	}

	printf("\033[2J");
}
#else
void process_bar_print(int len, const char *filename)
{
}
#endif


typedef struct{
	char *encap_cfg_file;
	unsigned char outer_vxlan_smac[6];
	unsigned short outer_vxlan_sport_net;
	unsigned char outer_vxlan_dmac[6];
	unsigned short outer_vxlan_dport_net;
	unsigned int outer_vxlan_sip_net;
	unsigned int outer_vxlan_dip_net;
	
}encap_args_t;

static encap_args_t g_vxlan_encap_args;


/* ascii字符转16进制 */
static char MESA_ascii_to_hex(char ascii)
{
	char c = 0;
	
	switch(ascii)
	{
		case '0':
		case '1':
		case '2':
		case '3':
		case '4':
		case '5':
		case '6':
		case '7':
		case '8':
		case '9':
			c = ascii - 0x30;
		break;

		case 'a':
		case 'b':
		case 'c':
		case 'd':
		case 'e':
		case 'f':
			c = 10 + ascii - 0x61;
		break;

		case 'A':
		case 'B':
		case 'C':
		case 'D':
		case 'E':
		case 'F':
			c = 10 + ascii - 0x41;
		break;
	}

	return c;
}

 /*  2012-04-11 LiJia add,将MAC字符串形式转换为16进制MAC地址.
参数:
	str:   MAC地址字符串
	delim: 字符串分隔符，常见为':', '-'等,如: xx:xx:xx:xx:xx:xx
			如果字符串无分隔符，delim设为-1.
	mac:   存储MAC地址的数组(指针),结果为网络序,
		   如网卡MAC地址为11:22:33:44:55:66,则mac[0]为0x11,mac[5]为0x66.
返回值:
	0: 正常
	-1:错误
*/
static int MESA_mac_pton(const char *str, int delim, char *mac)
{
#define MAC_STR_LEN_DELIM		(17) /* length of "11:22:33:44:55:66" */
#define MAC_STR_LEN_NODELIM		(12) /* length of "112233445566" */
	const char *s = str;
	int i; 

	/* 检查输入合法性 */
	if(delim != -1)
	{
		if(strlen(str) != MAC_STR_LEN_DELIM) 
		{
			printf("MAC string length error!\n");
			return -1;
		}
	}
	else
	{
		if(strlen(str) != MAC_STR_LEN_NODELIM) 
		{
			printf("MAC string length error!\n");
			return -1;
		}
	}

	/* 检查输入合法性，同时转换成16进制值 */
	for(i = 0; i < 6; i++)
	{
		mac[i] = 0; /* 先清零，赋值语句都是或操作 */
		if(isxdigit(*s)==0)
		{
			printf("MAC string type error!\n");
			return -1;
		}
		mac[i] |= MESA_ascii_to_hex(*s) << 4;
		s++;
		
		if(isxdigit(*s)==0)
		{
			printf("MAC string type error!\n");
			return -1;
		}
		mac[i] |= MESA_ascii_to_hex(*s);
		s++;
		
		if((delim != -1) && i<5 && (*s++ != (char)delim))
		{
			printf("MAC string type error!\n");
			return -1;
		}
	}

	return 0;
}

/*
[main]
vxlan_smac=74:86:7a:d0:12:fc
vxlan_dmac=00-01-6C-53-A9-94
vxlan_sip=10.0.6.201
vxlan_dip=10.0.6.59
vxlan_sport=12345
vxlan_dport=4789
*/
static int parse_encap_cfg_file(const char *filename)
{
	printf("parse encapsulation cfg file:%s\n", filename);
	int ret;
	unsigned char vxlan_smac[20], vxlan_dmac[20];
	unsigned char sip_str[16], dip_str[16];
	unsigned short tport;

	MESA_load_profile_string_def(options.encap_cfg_file, "main", "vxlan_smac", vxlan_smac, 20, "#");
	MESA_load_profile_string_def(options.encap_cfg_file, "main", "vxlan_dmac", vxlan_dmac, 20, "#");
	
	MESA_load_profile_string_def(options.encap_cfg_file, "main", "vxlan_sip", sip_str, 16, "#");
	MESA_load_profile_string_def(options.encap_cfg_file, "main", "vxlan_dip", dip_str, 16, "#");
	
	MESA_load_profile_short_def(options.encap_cfg_file, "main", "vxlan_sport", &tport, 0);
	g_vxlan_encap_args.outer_vxlan_sport_net = htons(tport);
	
	MESA_load_profile_short_def(options.encap_cfg_file, "main", "vxlan_dport", &tport, 0);
	g_vxlan_encap_args.outer_vxlan_dport_net = htons(tport);
	
	ret = MESA_mac_pton((char *)vxlan_smac, ':', g_vxlan_encap_args.outer_vxlan_smac);
	if(ret < 0){
		ret = MESA_mac_pton((char *)vxlan_smac, '-', g_vxlan_encap_args.outer_vxlan_smac);
		if(ret < 0){
			printf("config: %s->vxlan_smac invalid!\n", options.encap_cfg_file);
			return -1;
		}
	}	

	ret = MESA_mac_pton((char *)vxlan_dmac, ':', g_vxlan_encap_args.outer_vxlan_dmac);
	if(ret < 0){
		ret = MESA_mac_pton((char *)vxlan_dmac, '-', g_vxlan_encap_args.outer_vxlan_dmac);
		if(ret < 0){
			printf("config: %s->vxlan_dmac invalid!\n", options.encap_cfg_file);
			return -1;
		}
	}	

	ret = inet_pton(AF_INET, sip_str, &g_vxlan_encap_args.outer_vxlan_sip_net);
	if(ret <= 0){
		printf("config: %s->vxlan_sip invalid!\n", options.encap_cfg_file);
		return -1;
	}

	ret = inet_pton(AF_INET, dip_str, &g_vxlan_encap_args.outer_vxlan_dip_net);
	if(ret <= 0){
		printf("config: %s->vxlan_dip invalid!\n", options.encap_cfg_file);
		return -1;
	}	
	
	return 0;
}

/*************************** 内部实现接口 ********************************/

/*
 * Simple doubly linked list implementation.
 *
 * Some of the internal functions ("__xxx") are useful when
 * manipulating whole lists rather than single entries, as
 * sometimes we already know the next/prev entries and we can
 * generate better code by using them directly rather than
 * using the generic single-entry routines.
 */

static void INIT_LIST_HEAD(struct list_index *head)
{
	head->nextele = head;
	head->preele = head;
}


/*
 * Insert a new entry between two known consecutive entries.
 *
 * This is only for internal list manipulation where we know
 * the prev/next entries already!
 */
static inline void __list_add(struct list_index *new_list,
			      struct list_index *prev,
			      struct list_index *next)
{
	next->preele = new_list;
	new_list->nextele = next;
	new_list->preele = prev;
	prev->nextele = new_list;
}

/**
 * list_add - add a new entry
 * @new: new entry to be added
 * @head: list head to add it after
 *
 * Insert a new entry after the specified head.
 * This is good for implementing stacks.
 * lijia comment: list_add()把新节点插在head与head->next之间.
 */
void list_add(struct list_index *new_list, struct list_index *head)
{
	__list_add(new_list, head, head->nextele);
}


/**
 * list_add_tail - add a new entry
 * @new: new entry to be added
 * @head: list head to add it before
 *
 * Insert a new entry before the specified head.
 * This is useful for implementing queues.
 * lijia comment: list_add_tail()把新节点插在head与head->prev之间.
 */
void list_add_tail(struct list_index *new_list, struct list_index *head)
{
	__list_add(new_list, head->preele, head);
}

/*
 * Delete a list entry by making the prev/next entries
 * point to each other.
 *
 * This is only for internal list manipulation where we know
 * the prev/next entries already!
 */
static inline void __list_del(struct list_index * prev, struct list_index * next)
{
	next->preele = prev;
	prev->nextele = next;
}

static inline void __list_del_entry(struct list_index *entry)
{
	__list_del(entry->preele, entry->nextele);
	entry->nextele = NULL;
	entry->preele = NULL;
}

/**
 * list_del - deletes entry from list.
 * @entry: the element to delete from the list.
 * Note: list_empty() on entry does not return true after this, the entry is
 * in an undefined state.
 */

static void list_del(struct list_index *entry)
{
	__list_del(entry->preele, entry->nextele);
	entry->nextele = NULL;
	entry->preele = NULL;
}


/**
 * list_move - delete from one list and add as another's head
 * @list: the entry to move
 * @head: the head that will precede our entry
 */
static void list_move(struct list_index *list, struct list_index *head)
{
	__list_del_entry(list);
	list_add(list, head);
}

/**
 * list_move_tail - delete from one list and add as another's tail
 * @list: the entry to move
 * @head: the head that will follow our entry
 */
static void list_move_tail(struct list_index *list, struct list_index *head)
{
	__list_del_entry(list);
	list_add_tail(list, head);
}

#if 0
/**
 * list_empty - tests whether a list is empty
 * @head: the list to test.
 */
static int list_empty(const struct list_index *head)
{
	return head->nextele == head;
}
#endif

/**
 * list_empty_careful - tests whether a list is empty and not being modified
 * @head: the list to test
 *
 * Description:
 * tests whether a list is empty _and_ checks that no other CPU might be
 * in the process of modifying either member (next or prev)
 *
 * NOTE: using list_empty_careful() without synchronization
 * can only be safe if the only activity that can happen
 * to the list entry is list_del_init(). Eg. it cannot be used
 * if another CPU could re-list_add() it.
 */
static int list_empty_careful(const struct list_index *head)
{
	MESA_queue_head_t *next = head->nextele;
	return (next == head) && (next == head->preele);
}

static inline void list_count_init(void **list_count)
{
    long *p_c = (long *)list_count;
    *p_c = 0;
}

static inline void list_count_inc(void **list_count)
{
    long *p_c = (long *)list_count;
    long c = *p_c;
    *p_c = c + 1;
}

static inline void list_count_dec(void **list_count)
{
    long *p_c = (long *)list_count;
    long c = *p_c;
    *p_c = c - 1;
}

#if MESA_LIST_CHECK != 0
/*Function:Check the intergrity of the list,to dectect memory mess.
*Input: Queue Head,data check call back function
*Output: return a number below zero,if the queue got a problem,else return 0;
*Author:[email protected] at 20100913
*/
#if 0
static int MESA_q_list_check(const MESA_queue_head_t *qhead_obj,int(*quiddity_check)(const void *))
{
	MESA_list_index_t	*p=NULL, *head=NULL;
//	int linked_ele_number=0;
	int ret=0;
//	return 0;
	if(qhead_obj->listcount==0)
	{
		if(qhead_obj->head!=NULL||qhead_obj->head!=NULL)
		{
			ret=-1;
		}
		goto exit;
	}
	
	if(qhead_obj->head==NULL||qhead_obj->tail==NULL)
	{
		ret=-2;
		goto exit;
	}
	if(qhead_obj->listcount>1){
		if(qhead_obj->head->preele!=NULL||qhead_obj->head->nextele==NULL)
		{
			ret=-3;
			goto exit;
		}
		if(qhead_obj->tail->preele==NULL||qhead_obj->tail->nextele!=NULL)
		{
			ret=-4;
			goto exit;
		}

		head = p = qhead_obj->head;
		p = p->nextele;
		while(p != head)
		{
            if(p == NULL) break;
            if(p == head){
                ret = -5; /* has a cycle */
                goto exit;
            }
            p = p->nextele;
		}
	}
	/*
	pre=qhead_obj->head;
	p=pre->nextele;
	while(p!=NULL){
		linked_ele_number++;
		
		//Is the declared size equal to element linked number;
		if(linked_ele_number > qhead_obj->listcount)
		{
			ret=-5;
			goto exit;
		}
		
		//Is element's preele pointer equal to its preele
		if(pre!=p->preele)
		{
			ret=-6;
			goto exit;
		}
		if(quiddity_check!=NULL){
			if(0>quiddity_check(p->quiddity))
			{
				ret =-7;
				goto exit;
			}
		}
		pre=p;
		p=p->nextele;
	}
	//Is the last element equal to tail
	if(pre!=qhead_obj->tail)
	{
		ret=-8;
		goto exit;
	}
	if(linked_ele_number !=qhead_obj->listcount-1)
	{
		ret=-9;
		goto exit;
	}
	*/
exit:
	if(ret<0)
	{
		return ret;
	}
	else
	{
		return 1;
	}
	
}
#endif

#if 2==MESA_LIST_CHECK
/*Simple check,not raversal*/
static int MESA_q_is_item_in_list(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj){
//	MESA_list_index_t* pre=lindex_obj->preele;
//	MESA_list_index_t*next=lindex_obj->nextele;
	MESA_list_index_t*p=NULL;
	int i, num;
	
	if(list_empty_careful(qhead_obj)){
		return 0;
    }
	
	p = qhead_obj->nextele;
	num = MESA_get_list_count(qhead_obj);
    i = 0;
	while((p != qhead_obj) && (i <= num))
	{
		if(p == lindex_obj)
		{
			return 1;
		}
		p=p->nextele;
		i++;
	}
	
	return 0;
	
}
#endif

/*every MESA_list_index_t leaves list,its pre and next will be set NULL
 * In Configuration Transmiiter project, not null element will consider in list,
 * pre and next is NULL only if there's one element in list only*/
static int MESA_q_is_item_in_list_quick(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
	//empty list
	if(list_empty_careful(qhead_obj)){
		return 0;
	}	

	//have more element
	if(lindex_obj->nextele==NULL || lindex_obj->preele==NULL){
		return 0;
	}
	if(lindex_obj->nextele->preele != lindex_obj){
        return 0;
	}
    if(lindex_obj->preele->nextele != lindex_obj){
        return 0;
	}

	return 1;

}
#endif

/*************************** 外部调用接口 ********************************/


/*
    第一次使用MESA_list模块时，必须调用此初始化模块.
*/
int MESA_list_init(MESA_queue_head_t *head)
{
    INIT_LIST_HEAD(head);
    list_count_init((void **)&head->quiddity);
    return 0;
}

/* 做为head使用时, "quiddity"做为一个long型变量,用于存储链表元素数量 */
long MESA_get_list_count(const MESA_queue_head_t *head)
{
    return (long)head->quiddity;
}

MESA_list_index_t *MESA_q_read_head(const MESA_queue_head_t *qhead_obj)
{
    if(list_empty_careful(qhead_obj)){
        return NULL;
    }
    return qhead_obj->nextele;
}


MESA_list_index_t *MESA_q_get_head(MESA_queue_head_t *qhead_obj)
{
    MESA_list_index_t *out;
    if(list_empty_careful(qhead_obj)){
        return NULL;
    }
    out = qhead_obj->nextele;
    list_del(out);
    list_count_dec((void **)&qhead_obj->quiddity);

    return out;
}

MESA_list_index_t *MESA_q_get_tail(MESA_queue_head_t *qhead_obj)
{
    MESA_list_index_t *out;
    if(list_empty_careful(qhead_obj)){
        return NULL;
    }
    out = qhead_obj->preele;
    list_del(out);
    list_count_dec((void **)&qhead_obj->quiddity);
    
    return out;
}

MESA_list_index_t *MESA_q_join_tail(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
    list_add_tail(lindex_obj, qhead_obj);
    list_count_inc((void **)&qhead_obj->quiddity);
    
    return lindex_obj;
}

MESA_list_index_t *MESA_q_join_head(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
    list_add(lindex_obj, qhead_obj);
    list_count_inc((void **)&qhead_obj->quiddity);
    
    return lindex_obj;
}

MESA_list_index_t *MESA_q_leave_list(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
#if 1==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list_quick(qhead_obj, lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#elif 2==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list(qhead_obj, lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#endif    
    list_del(lindex_obj);
    list_count_dec((void **)&qhead_obj->quiddity);

    return lindex_obj;
}

MESA_list_index_t *MESA_q_move_head(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
#if 1==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list_quick(qhead_obj, lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#elif 2==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list(qhead_obj, lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#endif    
    list_move(lindex_obj, qhead_obj);

    return lindex_obj;
}

MESA_list_index_t *MESA_q_move_tail(MESA_queue_head_t *qhead_obj, MESA_list_index_t *lindex_obj)
{
#if 1==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list_quick(qhead_obj,lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#elif 2==MESA_LIST_CHECK
    if(0 == MESA_q_is_item_in_list(qhead_obj, lindex_obj)){
        //return NULL;
        assert(0); //critical
    }
#endif    
    list_move_tail(lindex_obj, qhead_obj);

    return lindex_obj;
}

/************************ MESA_fixed_q implement *************************/
/************************ MESA_fixed_q 内部实现接口***********************/
MESA_fixed_q_t *__create_fixed_q_elem(long max_elem_len)
{    
    MESA_fixed_q_t *tq;
    MESA_fixed_qelem_t *tqelem;
    void *tdata;
    
    tq = (MESA_fixed_q_t *)malloc(sizeof(MESA_fixed_q_t));
    memset(tq, 0, sizeof(MESA_fixed_q_t));

    tdata = malloc(max_elem_len);

    tqelem = (MESA_fixed_qelem_t *)malloc(sizeof(MESA_fixed_qelem_t));
    tqelem->data = tdata;
    tqelem->datalen = 0;

    tq->quiddity = (void *)tqelem;

    return tq;
}


/************************ MESA_fixed_q 外部接口 **************************/
/*
    第一次使用MESA_fixed模块时，必须调用此初始化模块.
    total_elem_num: total queue element number;
    max_elem_len  : max length of a queue element.
*/
int MESA_fixed_q_init(MESA_fixed_q_t *__head, long total_elem_num, long max_elem_len)
{
    long i;
    MESA_fixed_q_t *real_head, *real_tail;
    MESA_fixed_qelem_t *tqelem;

    if(!__head || (0 >= total_elem_num) || (0 >= max_elem_len)){
        return -1;
    }

    __head->quiddity = malloc(sizeof(MESA_fixed_qelem_t));

    /* 第一个节点形成自环的双向链表 */
    real_head = __create_fixed_q_elem(max_elem_len);
    INIT_LIST_HEAD(real_head);

    /* 后续的节点插入链表 */    
    for(i = 0; i < total_elem_num; i++){
        real_tail = __create_fixed_q_elem(max_elem_len);
        list_add_tail(real_tail, real_head);
    }

    __head->nextele = (struct list_index *)real_head; /* 头节点.nextele用于存储当前队列的实际队头节点指针 */
    __head->preele = (struct list_index *)real_head;  /* 头节点.preele用于存储当前队列的实际队尾节点指针 */
    tqelem = (MESA_fixed_qelem_t *)__head->quiddity;
    list_count_init(&tqelem->data);  /* 头节点.data用于存储当前队列的实际元素数 */
    tqelem->datalen = max_elem_len;  /* 头节点.datalen用于存储当前队列的元素最大长度 */

    return 0;
}

long MESA_fixed_q_count(const MESA_fixed_q_t *__head)
{
    const MESA_fixed_qelem_t *tqelem = (MESA_fixed_qelem_t *)__head->quiddity;
    return (long)tqelem->data;
}

MESA_fixed_qelem_t *MESA_fixed_q_read_head(MESA_fixed_q_t *__head)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *real_tail = (MESA_fixed_q_t *)__head->preele;
    
    if(list_empty_careful(real_head)){
        return NULL;
    }
    
    if(MESA_fixed_q_count(__head) == 0){ /* queue is empty */
        return NULL;
    }
    
    return (MESA_fixed_qelem_t *)(real_head->quiddity);
}

MESA_fixed_qelem_t *MESA_fixed_q_get_head(MESA_fixed_q_t *__head)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *real_tail = (MESA_fixed_q_t *)__head->preele;
    MESA_fixed_qelem_t *tq, *ret;
    
    if(list_empty_careful(real_head)){
        return NULL;
    }

    if(MESA_fixed_q_count(__head) == 0){ /* queue is empty */
        return NULL;
    }

    tq = (MESA_fixed_qelem_t *)__head->quiddity;
    list_count_dec(&tq->data);

    ret = (MESA_fixed_qelem_t *)real_head->quiddity;
    __head->nextele = real_head->nextele;

    return ret;
}

#if 0
const MESA_fixed_qelem_t *MESA_fixed_q_get_tail(MESA_fixed_q_t *__head)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *real_tail = (MESA_fixed_q_t *)__head->preele;
    MESA_fixed_qelem_t *tq, *ret;
    
    if(list_empty_careful(real_head)){
        return NULL;
    }

    if(real_head == real_tail){ /* queue is empty */
        return NULL;
    }

    tq = (MESA_fixed_qelem_t *)__head->quiddity;
    list_count_dec(&tq->data);

    ret = (MESA_fixed_qelem_t *)real_tail->quiddity;
    __head->preele = real_tail->preele;

    return ret;
}


const MESA_fixed_qelem_t *MESA_fixed_q_join_head(MESA_fixed_q_t *__head, void *data, long datalen)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *real_tail = (MESA_fixed_q_t *)__head->preele;
    MESA_fixed_qelem_t *tq, *ret;
    MESA_fixed_q_t *join_place;

    if((real_head->preele == real_tail)
     ||(real_tail->nextele == real_head)){ /* queue is full */
        return NULL;
    }

    tq = (MESA_fixed_qelem_t *)__head->quiddity;
    if(tq->datalen < datalen){
        return NULL;
    }
    list_count_inc(&tq->data);
    
    join_place = real_head->preele;
    ret = (MESA_fixed_qelem_t *)join_place->quiddity;
    memcpy(ret->data, data, datalen);
    ret->datalen = datalen;

    __head->nextele = join_place; /* update head pointer */

    return ret;
}
#endif

MESA_fixed_qelem_t *MESA_fixed_q_join_tail(MESA_fixed_q_t *__head, void *data, long datalen)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *real_tail = (MESA_fixed_q_t *)__head->preele;
    MESA_fixed_qelem_t *tq, *ret;
    MESA_fixed_q_t *join_place;

    tq = (MESA_fixed_qelem_t *)__head->quiddity;

    if((real_head->preele == real_tail)
     ||(real_tail->nextele == real_head)){ /* queue is full */
        return NULL;
    }

    if(tq->datalen < datalen){
        return NULL;
    }
    list_count_inc(&tq->data);
    
    join_place = real_tail->nextele;
    ret = (MESA_fixed_qelem_t *)join_place->quiddity;
    memcpy(ret->data, data, datalen);
    ret->datalen = datalen;

    __head->preele = join_place; /* update tail pointer */

    return ret;
}

void MESA_fixed_q_destroy(MESA_fixed_q_t *__head)
{
    MESA_fixed_q_t *real_head = (MESA_fixed_q_t *)__head->nextele;
    MESA_fixed_q_t *del;
    MESA_fixed_qelem_t *tq;

    while(!list_empty_careful(real_head)){
        del = real_head->nextele;
        list_del(del);
        tq = (MESA_fixed_qelem_t *)del->quiddity;
        free(tq->data);
        free(tq);
        free(del);
    }
    tq = (MESA_fixed_qelem_t *)real_head->quiddity;
    free(tq->data);
    free(tq);
    free(real_head);

    free(__head->quiddity);

    return;
}

#ifdef __cplusplus
}
#endif


MESA_fixed_q_t *pkt_queue;


#ifdef MARSIO
#include "marsio.h"
static struct mr_instance * tcpburst_marsio4_instance;
static struct mr_vdev *tcpburst_marsio4_vdev;
static struct mr_sendpath *tcpburst_marsio4_sendpath;
static char *marsio_send_device;
long marsio_cpu_mask = 0x02;

int stream_burst_marsio_init(void)
{
	char app_name[64];
	char *cpu_msk_str;

#ifndef TCPREPLAY_EDIT
	cpu_msk_str = OPT_VALUE_CPU_MASK;
	if(strncasecmp("0x", cpu_msk_str, 2) != 0){
		fprintf(stderr,"%s\n","cpu-mask arg invalid, for example: 0xF0!\n");
		exit(1);
	}
	marsio_cpu_mask = strtoul(cpu_msk_str, NULL, 16);
#endif

	tcpburst_marsio4_instance = marsio_create();
	if(NULL == tcpburst_marsio4_instance) {
		fprintf(stderr,"%s\n","marsio_create() error!\n");
		exit(1);
	}

	int marsio_opt = 1;
	marsio_option_set(tcpburst_marsio4_instance, MARSIO_OPT_THREAD_NUM, &marsio_opt, sizeof(int));

	marsio_option_set(tcpburst_marsio4_instance, MARSIO_OPT_THREAD_MASK, &marsio_cpu_mask, sizeof(long));

	srand(time(NULL));
	//snprintf(app_name, 64, "%s_%c%c%c", "tcpburst", 'a'+random()%26, 'a'+random()%26,'a'+random()%26);
	snprintf(app_name, 64, "%s_%d", "tcpburst", getpid());
	
	if(marsio_init(tcpburst_marsio4_instance, app_name) < 0){
		fprintf(stderr,"%s\n","marsio_init() error!\n");
		exit(1);
	}

	tcpburst_marsio4_vdev = marsio_open_device(tcpburst_marsio4_instance, marsio_send_device, 0/* rx_stream */, 1 /* tx_stream */);
	if(NULL == tcpburst_marsio4_vdev) {
		fprintf(stderr,"%s\n","marsio_open_device %s error!\n", marsio_send_device);
		exit(1);
	}

	tcpburst_marsio4_sendpath = marsio_sendpath_create_by_vdev(tcpburst_marsio4_vdev);
	if(NULL == tcpburst_marsio4_sendpath) {
		fprintf(stderr,"%s\n","marsio_sendpath_create_by_vdev() error!\n");
		exit(1);
	}

	return 0;
}
#endif

static int stream_burst_init(void)
{
	int i;

	pkt_queue = (MESA_fixed_q_t *)malloc(options.stream_multiple * sizeof(MESA_fixed_q_t));
	memset(pkt_queue, 0, options.stream_multiple * sizeof(MESA_fixed_q_t));

	if(options.pkt_distance > 0)
	{
		for(i = 0; i < options.stream_multiple; i++)
		{
			MESA_fixed_q_init(&pkt_queue[i], options.pkt_distance*(i+1), TCP_BURST_MTU);
		}
	}

#ifdef MARSIO
	if(12 == send_pkt_driver_mode){
		stream_burst_marsio_init();
	}
#endif

#if PROCESS_BAR_SW
	process_bar_prt_init();
#endif
	return 0;
}


/*
	return value:
	1:  src addr(ip or port) bigger than dst addr;
	-1: src addr(ip or port) smaller than dst addr;
*/
static inline int stream_addr_cmp_tcp(struct ip *iphdr, struct tcphdr *tcphdr)
{
	int ret;
	
	if(iphdr->ip_src.s_addr > iphdr->ip_dst.s_addr)
	{
		ret = 1;
	}
	else if(iphdr->ip_src.s_addr < iphdr->ip_dst.s_addr)
	{
		ret = -1;
	}
	else
	{
		if(tcphdr->th_sport > tcphdr->th_dport)
		{
			ret = 1;
		}
		else
		{
			ret = -1;
		}
	}
	
	return ret;
}

/*
	return value:
	1:  src addr(ip or port) bigger than dst addr;
	-1: src addr(ip or port) smaller than dst addr;
*/
static inline int stream_addr_cmp_udp(struct ip *iphdr, struct udphdr *udphdr)
{
	int ret;
	
	if(iphdr->ip_src.s_addr > iphdr->ip_dst.s_addr)
	{
		ret = 1;
	}
	else if(iphdr->ip_src.s_addr < iphdr->ip_dst.s_addr)
	{
		ret = -1;
	}
	else
	{
		if(udphdr->uh_sport > udphdr->uh_dport)
		{
			ret = 1;
		}
		else
		{
			ret = -1;
		}
	}
	
	return ret;
}


/* 修改原包的IP地址，使四元组和原包不同.
   为了不从新计算校验和，以达到较高的发送速率，
   改地址时，仅对IP地址进行修改，增加和减少的值要相等.
*/
static int stream_edit_addr(struct ip* raw_iphdr, int differ, struct ip* iphdr)
{
	//struct ip *iphdr;
	struct tcphdr *tcphdr;
	struct udphdr *udphdr;

	differ *= 3;

	//iphdr = (struct ip *)(pkt + 14);

	if(IPPROTO_TCP == raw_iphdr->ip_p || IPPROTO_UDP == raw_iphdr->ip_p)
	{
		tcphdr = (struct tcphdr *)((char*)iphdr + iphdr->ip_hl*4);
		if(stream_addr_cmp_tcp(raw_iphdr, tcphdr) < 0)
		{		
			iphdr->ip_src.s_addr -= differ;
			iphdr->ip_dst.s_addr += differ;
		}
		else
		{
			iphdr->ip_src.s_addr += differ;
			iphdr->ip_dst.s_addr -= differ;
		}
	}else{ /* 增加对GRE, IPinIP, 6over4的支持 */
		if(iphdr->ip_src.s_addr <= iphdr->ip_dst.s_addr){
			iphdr->ip_src.s_addr += differ;
			iphdr->ip_dst.s_addr -= differ;
		}else{
			iphdr->ip_src.s_addr -= differ;
			iphdr->ip_dst.s_addr += differ;
		}
	}
	
	return 0;
}

#ifdef MARSIO
static int stream_burst_send_pkt_by_marsio(struct mr_sendpath *handle, const u_char *pkt, size_t pktlen)
{
	marsio_buff_t *send_mbuf[1];
	char *real_buf;
	int ret;

	if(pktlen > TCP_BURST_MTU){
		return -1;
	}

	ret = marsio_buff_malloc_global(tcpburst_marsio4_instance, send_mbuf, 1, MARSIO_SOCKET_ID_ANY, 0);
	if(ret < 0){
		return -1;
	}	

	/* 此处使用append而不是mtod, append内部实际包括了set datalen的操作 */
	real_buf = marsio_buff_append(send_mbuf[0], pktlen);
	
	memcpy(real_buf, pkt, pktlen);

	marsio_send_burst(tcpburst_marsio4_sendpath, 0, send_mbuf, 1);
	
	return pktlen; 
}
#endif

//#ifdef MARSIO
#if 0
static stream_burst_send_pkt_multiple(void *handle, const u_char *pkt, size_t pktlen)
{
	marsio_buff_t *send_mbuf[256];
	char *real_buf;
	int i, ret;

	ret = marsio_buff_malloc_global(tcpburst_marsio4_instance, send_mbuf, 1, MARSIO_SOCKET_ID_ANY, 0);
	if(ret < 0){
		return -1;
	}	
	real_buf = marsio_buff_append(send_mbuf[0], pktlen);
	memcpy(real_buf, pkt, pktlen); /* 只copy一次数据 */
			
	for(i = 1; i < options.stream_multiple; i++){
		send_mbuf[i] = marsio_buff_clone_deep(tcpburst_marsio4_instance, send_mbuf[0], MARSIO_SOCKET_ID_ANY, 0);
		real_buf = marsio_buff_mtod(send_mbuf[i]);
		stream_edit_addr(real_buf, i + 1);
	}

	marsio_send_burst(tcpburst_marsio4_sendpath, 0, send_mbuf, options.stream_multiple);

#if 0
	if(options.stream_multiple > 1){
			real_buf = marsio_buff_append(send_mbuf[0], pktlen);
			memcpy(real_buf, pkt, pktlen); /* 只copy一次数据 */
			stream_edit_addr(real_buf, i + 1);
			marsio_send_burst_with_options(tcpburst_marsio4_sendpath, 0, send_mbuf, 1, MARSIO_SEND_OPT_NO_FREE);
			marsio_buff_reset(send_mbuf[0]);
			
		for(i = 1; i < options.stream_multiple; i++){
			/* 此处使用append而不是mtod, append内部实际包括了set datalen的操作 */
			real_buf = marsio_buff_append(send_mbuf[0], pktlen);
			stream_edit_addr(real_buf, i + 1);
			marsio_send_burst_with_options(tcpburst_marsio4_sendpath, 0, send_mbuf, 1, MARSIO_SEND_OPT_NO_FREE);
			marsio_buff_reset(send_mbuf[0]);
		}
		marsio_buff_free(tcpburst_marsio4_instance, send_mbuf, 1, MARSIO_SOCKET_ID_ANY, 0);
	}else{
		/* 此处使用append而不是mtod, append内部实际包括了set datalen的操作 */
		real_buf = marsio_buff_append(send_mbuf[0], pktlen);
		
		memcpy(real_buf, pkt, pktlen);

		marsio_send_burst(tcpburst_marsio4_sendpath, 0, send_mbuf, 1);
	}
#endif

	return pktlen; 
}
#endif

typedef struct{
	unsigned char flags;

	/*------------byte delim -------*/
	unsigned char reserved[3];

	/*--------int delim -------*/
	unsigned char vlan_id_half_high;
	unsigned char link_layer_type : 4; /* 二层报文封装格式 */
	unsigned char vlan_id_half_low : 4;
	unsigned int online_test : 1;
	unsigned int link_id : 6;
	unsigned int dir : 1;

	unsigned int r7 : 1;
	unsigned int r6 : 1;
	unsigned int r5 : 1;
	unsigned int r4 : 1;
	unsigned int vni_flag : 1;
	unsigned int r2 : 1;
	unsigned int r1 : 1;
	unsigned int r0 : 1;

}tcpburst_vxlan_hdr_t;

#include <linux/if_ether.h>
#include <netinet/ip.h>
#include <netinet/udp.h>
static int sendpacket_with_encapsulation(sendpacket_t *handle, const u_char *pkt, size_t pktlen)
{
	char encap_pkt_buf[2048];
	int ret;
	
	struct ethhdr *ehdr = (struct ethhdr *)encap_pkt_buf;
	memcpy(ehdr->h_source, g_vxlan_encap_args.outer_vxlan_smac, 6);
	memcpy(ehdr->h_dest, g_vxlan_encap_args.outer_vxlan_dmac, 6);
	ehdr->h_proto = ntohs(ETH_P_IP);

	struct ip *ihdr = (struct ip *)(encap_pkt_buf + sizeof(struct ethhdr));
	ihdr->ip_v = 4;
	ihdr->ip_hl = 5;
	ihdr->ip_tos = 0;
	ihdr->ip_len = htons(sizeof(struct ip) + sizeof(struct udphdr) + sizeof(tcpburst_vxlan_hdr_t) + pktlen);
	ihdr->ip_id = 0x3412;
	ihdr->ip_off = 0;
	ihdr->ip_ttl = 64;
	ihdr->ip_p = 17;
	ihdr->ip_sum = 0;
	ihdr->ip_src.s_addr = g_vxlan_encap_args.outer_vxlan_sip_net;
	ihdr->ip_dst.s_addr = g_vxlan_encap_args.outer_vxlan_dip_net;

	struct udphdr *uhdr = (struct udphdr *)(encap_pkt_buf + sizeof(struct ethhdr) + sizeof(struct ip));

	uhdr->uh_sport = g_vxlan_encap_args.outer_vxlan_sport_net;
	uhdr->uh_dport = g_vxlan_encap_args.outer_vxlan_dport_net;
	uhdr->uh_ulen = htons(sizeof(struct udphdr) + sizeof(tcpburst_vxlan_hdr_t) + pktlen);
	uhdr->uh_sum = 0;

	tcpburst_vxlan_hdr_t *vxlan_hdr = (tcpburst_vxlan_hdr_t *)((char *)uhdr + sizeof(struct udphdr));

	vxlan_hdr->flags = 0;
	vxlan_hdr->reserved[0] = 0;
	vxlan_hdr->reserved[1] = 0;
	vxlan_hdr->reserved[2] = 0;
	vxlan_hdr->link_id = 101;

	memcpy(encap_pkt_buf + sizeof(struct ethhdr) + sizeof(struct ip) + sizeof(struct udphdr) + sizeof(tcpburst_vxlan_hdr_t), pkt, pktlen);

	return sendpacket(handle, encap_pkt_buf, pktlen + sizeof(struct ethhdr) + sizeof(struct ip) + sizeof(struct udphdr) + sizeof(tcpburst_vxlan_hdr_t));
}

int stream_burst_send_pkt(void *handle, const u_char *pkt, size_t pktlen)
{
	if(1 == send_pkt_driver_mode){
		if(options.encap_cfg_file != NULL){
			return sendpacket_with_encapsulation((sendpacket_t *)handle, pkt, pktlen);
		}
		return sendpacket((sendpacket_t *)handle, pkt, pktlen);
	}

#ifdef MARSIO
	return stream_burst_send_pkt_by_marsio((struct mr_sendpath *)handle, pkt, pktlen);
#endif
}

struct mesa_mpls_hdr
{
#if __BYTE_ORDER == __LITTLE_ENDIAN
	unsigned short mpls_label_low;
	unsigned char mpls_bls : 1; /* bottom of label stack */
	unsigned char mpls_exp : 3;
	unsigned char mpls_label_high : 4;
	unsigned char mpls_ttl;
#elif __BYTE_ORDER == __BIG_ENDIAN
	unsigned char mpls_ttl;
	unsigned char mpls_label_high : 4;
	unsigned char mpls_exp : 3;
	unsigned char mpls_bls : 1; /* bottom of label stack */
	unsigned short mpls_label_low;
#else
#error "Please check <endian.h>"
#endif
};

struct mesa_pppoe_session_hdr
{
#if __BYTE_ORDER == __LITTLE_ENDIAN
	unsigned int ver : 4;
	unsigned int type : 4;
#elif __BYTE_ORDER == __BIG_ENDIAN
	unsigned int type : 4;
	unsigned int ver : 4;
#else
#error "Please check <endian.h>"
#endif
	unsigned char code;
	unsigned short session_id;
	unsigned short len;
	/* to do:
		ppp应该单独作为一层存在, 为了简化处理, 强制将其和PPPOE_SES绑在一起,
		如果需要监听PPP协商过程, 此结构需要改动.
	*/
	unsigned short ppp_protocol;
} __attribute__((packed, aligned(1)));

#define PPP_PROTOCOL_IPv4 (0x0021)

static const u_char *find_iphdr(const u_char *pkt, size_t pktlen, const u_char *this_layer_data, u_short this_layer_type)
{
	if (pkt == NULL || pktlen <= 0 || this_layer_data == NULL)
	{
		return NULL;
	}
	if(this_layer_data - pkt >= pktlen)
	{
		return NULL;
	}
	u_short proto_type = 0;
	struct mesa_mpls_hdr *this_mpls_hdr = NULL;
	struct mesa_pppoe_session_hdr * ppp_hdr= NULL;

	switch (this_layer_type)
	{
	
		case ETH_P_IP:
		case ETH_P_IPV6:
			 return this_layer_data;

		case ETH_P_8021Q:
			proto_type = *((unsigned short *)((char *)this_layer_data + 2));
			return find_iphdr(pkt, pktlen, this_layer_data + 4, ntohs(proto_type));

		case ETH_P_MPLS_UC:
			this_mpls_hdr = (struct mesa_mpls_hdr *)this_layer_data;
			if (1 != this_mpls_hdr->mpls_bls)
			{
				return find_iphdr(pkt, pktlen, this_layer_data + 4, ETH_P_MPLS_UC);
			}
			else
			{
				return this_layer_data+4;
			}

		case ETH_P_PPP_SES:
			ppp_hdr = (struct mesa_pppoe_session_hdr *)this_layer_data;
			if (PPP_PROTOCOL_IPv4 == ntohs(ppp_hdr->ppp_protocol))
			{
				return this_layer_data+8;
			}
		default:
			break;
	}
	return NULL;
}

	int stream_burst(sendpacket_t *sp, const u_char *pkt, size_t pktlen, int flush, int cache_file_idx)
{
	u_char *pdata;
	MESA_fixed_qelem_t *q_obj;
	int i;
	unsigned char pbuf[TCP_BURST_MTU];

	if(cache_file_idx != dumpfile_last_file_index){
		dumpfile_last_file_index = cache_file_idx;
		dumpfile_total_size = __get_file_length(options.files[cache_file_idx]);
	}

	process_bar_print(pktlen, options.files[cache_file_idx]);

	//struct ip *iphdr = (struct ip *)(pkt + 14);
	struct ethhdr * eth = (struct ethhdr *)pkt;
	struct ip *iphdr = NULL;
	struct ip *t_iphdr = NULL;
	
	if(1 == flush) /* 原始包已读完, 强制刷新队列中剩余数据包 */
	{
		if(options.pkt_distance > 0)
		{
			/* 取出之前的包发送 */
			for(i = 0; i < options.stream_multiple; i++)
			{	
				while((q_obj = MESA_fixed_q_get_head(&pkt_queue[i])) != NULL)
				{
					stream_burst_send_pkt(sp, q_obj->data, q_obj->datalen);
					pkts_sent ++;
	        			bytes_sent += pktlen;
				}
			}
		}
		return 0;
	}

	if(pkt != NULL && pktlen > 14)
	{
		iphdr=find_iphdr(pkt, pktlen, (pkt + 14), ntohs(eth->h_proto));
	}

	if (iphdr == NULL)
	{
		if ((options.stream_multiple != 0) && (options.verbose != 0))
		{
			fprintf(stderr, "Not support encap pkt, cannot find iphdr%d\n");
		}
		return -1;
	}


	/* 支持TCP和UDP, GRE, IPinIP协议 */
	if((iphdr->ip_p != IPPROTO_TCP) 
	 && (iphdr->ip_p != IPPROTO_UDP) 
	 && (iphdr->ip_p != IPPROTO_GRE)
	 && (iphdr->ip_p != IPPROTO_IPIP)
	 && (iphdr->ip_p != IPPROTO_IPV6))
	{
		if((options.stream_multiple != 0) && (options.verbose != 0)){
			fprintf(stderr, "Not support amplify protocol:%d\n", iphdr->ip_p);
		}
		return -1; 
	}

	if(pktlen > TCP_BURST_MTU || pktlen > options.max_burst_pkt_len || pktlen < options.min_burst_pkt_len){
		//fprintf(stderr, "Tcpburst error! Packet too long:%d, current MTU is:%d\n", pktlen, TCP_BURST_MTU);
		return -1;
	}
	
	if(options.stream_multiple > 0)
	{
		memcpy(pbuf, pkt, pktlen);
		t_iphdr = (struct ip *)(pbuf+((const u_char*)iphdr - pkt));	
	}
	if(options.pkt_distance > 0)
	{
		/* 暂存到队列 */
		for(i = 0; i < options.stream_multiple; i++)
		{
			stream_edit_addr(iphdr, i+1, t_iphdr);
			MESA_fixed_q_join_tail(&pkt_queue[i], pbuf, pktlen);
		}

		/* 取出之前的包发送 */
		for(i = 0; i < options.stream_multiple; i++)
		{
			if(MESA_fixed_q_count(&pkt_queue[i]) >= options.pkt_distance*(i+1))
			{		
				q_obj = MESA_fixed_q_get_head(&pkt_queue[i]);
				stream_burst_send_pkt(sp, q_obj->data, q_obj->datalen);
				pkts_sent ++;
        			bytes_sent += pktlen;
			}
		}
	}
	else
	{
		/* 如果没有包间距需求, 直接发送N倍数据包即可 */
		for(i = 0; i < options.stream_multiple; i++)
		{
			stream_edit_addr(iphdr, i+1, t_iphdr); /* 在函数内部修改包头IP */
			stream_burst_send_pkt(sp, pbuf, pktlen);
			pkts_sent ++;
    			bytes_sent += pktlen;
		}
	}

	return 0;
}

void stream_burst_destroy(void)
{
	int i;

	if(0 == options.pkt_distance)
	{
		return;
	}

	for(i = 0; i < options.stream_multiple; i++)
	{
		MESA_fixed_q_destroy(&pkt_queue[i]);
	}

	free(pkt_queue);
}

#endif

#ifdef MARSIO
static void *replay_files_thread(void *arg)
{
	int i;
	int loop = (int)arg;

	marsio_thread_init(tcpburst_marsio4_instance);

	if(0 == loop){
		while(1)replay_file(0);
	}
	for(i = 0; i < loop; i++){
		replay_file(0);
	}

	return NULL;
}
#endif

int
main(int argc, char *argv[])
{
    int i, optct = 0;
#ifdef TCPREPLAY_EDIT
    int rcode;
#endif

    init();                     /* init our globals */

    optct = optionProcess(&tcpreplayOptions, argc, argv);
    argc -= optct;
    argv += optct;

#ifdef TCPBURST
#ifndef TCPREPLAY_EDIT
	options.driver_mode = OPT_VALUE_DRIVER_MODE;
	if(NULL == options.driver_mode){
		options.driver_mode = "pcap"; /* 没指定参数, default is pcap */
	}
#ifdef MARSIO	
    if(strncasecmp(options.driver_mode, "marsio", 6) == 0){
	send_pkt_driver_mode = 12;
	marsio_send_device = get_interface(NULL, OPT_ARG(INTF1)); /* 获取原始-i输出的接口名称 */
	OPT_ARG(INTF1) = strdup("lo"); /* 使用lo接口骗过tcpreplay跟pcap相关的初始化 */
    }	else
#endif    
    if(strncasecmp(options.driver_mode, "pcap", 6) == 0){
	send_pkt_driver_mode = 1;
    }else{
    	printf("Invalid driver-mode:%s\n", options.driver_mode);
	exit(1);
    }
#endif
#endif

    post_args();

#ifdef TCPBURST
#ifndef TCPREPLAY_EDIT
	stream_burst_init();
#endif
#endif

#ifdef TCPREPLAY_EDIT
    /* init tcpedit context */
    if (tcpedit_init(&tcpedit, sendpacket_get_dlt(options.intf1)) < 0) {
        errx(-1, "Error initializing tcpedit: %s", tcpedit_geterr(tcpedit));
    }

    /* parse the tcpedit args */
    rcode = tcpedit_post_args(&tcpedit);
    if (rcode < 0) {
        errx(-1, "Unable to parse args: %s", tcpedit_geterr(tcpedit));
    } else if (rcode == 1) {
        warnx("%s", tcpedit_geterr(tcpedit));
    }

    if (tcpedit_validate(tcpedit) < 0) {
        errx(-1, "Unable to edit packets given options:\n%s",
                tcpedit_geterr(tcpedit));
    }
#endif

    if ((options.enable_file_cache || options.preload_pcap) && ! HAVE_OPT(QUIET)) {
        notice("File Cache is enabled");
    }

    /*
     * Setup up the file cache, if required
     */
    if (options.enable_file_cache || options.preload_pcap) {
        options.file_cache = safe_malloc(argc * sizeof(file_cache_t));

        /*
         *  Initialise each of the file cache structures
         */
        for (i = 0; i < argc; i++) {
            options.file_cache[i].index = i;
            options.file_cache[i].cached = FALSE;
            options.file_cache[i].packet_cache = NULL;
        }
    }

    for (i = 0; i < argc; i++) {
        options.files[i] = safe_strdup(argv[i]);

        /* preload our pcap file? */
        if (options.preload_pcap) {
            preload_pcap_file(i);
        }
    }

    /* init the signal handlers */
    init_signal_handlers();

    if (gettimeofday(&begin, NULL) < 0)
        errx(-1, "gettimeofday() failed: %s",  strerror(errno));


	if (12 == send_pkt_driver_mode)
	{	
#ifdef MARSIO		
		pthread_t pid;
		pthread_create(&pid, NULL, replay_files_thread, (void *)options.loop);
		pthread_join(pid, NULL);
#else
		errx(-1, "send_pkt_driver_mode == %s, but tcp_burst did not compiled with %s", "MARSIO", "MARSIO");
#endif		
	}
	else
	{
		/* main loop for non-bridge mode */
		if (options.loop > 0) {
			while (options.loop--) {  /* limited loop */
				/* process each pcap file in order */
				for (i = 0; i < argc; i++) {
					/* reset cache markers for each iteration */
					cache_byte = 0;
					cache_bit = 0;
					replay_file(i);               
				}
			}
		}
		else {
			/* loop forever */
			while (1) {
				for (i = 0; i < argc; i++) {
					/* reset cache markers for each iteration */
					cache_byte = 0;
					cache_bit = 0;              
					replay_file(i);           
				}
			}
		}
	}
    if (bytes_sent > 0) {
        if (gettimeofday(&end, NULL) < 0)
            errx(-1, "Unable to gettimeofday(): %s", strerror(errno));

        packet_stats(&begin, &end, bytes_sent, pkts_sent, failed);
        
        if (options.intf1 != NULL)
            printf("%s", sendpacket_getstat(options.intf1));
        if (options.intf2 != NULL)
            printf("%s", sendpacket_getstat(options.intf2));
    }

#ifdef TCPBURST
#ifndef TCPREPLAY_EDIT
	stream_burst_destroy();
#endif
#endif

    return 0;
}   /* main() */

/**
 * \brief Preloads the memory cache for the given pcap file_idx 
 *
 * Preloading can be used with or without --loop and implies using
 * --enable-file-cache
 */
void
preload_pcap_file(int file_idx)
{
    char *path = options.files[file_idx];
    pcap_t *pcap = NULL;
    char ebuf[PCAP_ERRBUF_SIZE];
    const u_char *pktdata = NULL;
    struct pcap_pkthdr pkthdr;
    packet_cache_t *cached_packet = NULL;
    packet_cache_t **prev_packet = &cached_packet;
    COUNTER packetnum = 0;

    /* close stdin if reading from it (needed for some OS's) */
    if (strncmp(path, "-", 1) == 0)
        if (close(1) == -1)
            warnx("unable to close stdin: %s", strerror(errno));

    if ((pcap = pcap_open_offline(path, ebuf)) == NULL)
        errx(-1, "Error opening pcap file: %s", ebuf);

#ifdef HAVE_PCAP_SNAPSHOT
    if (pcap_snapshot(pcap) < 65535)
        warnx("%s was captured using a snaplen of %d bytes.  This may mean you have truncated packets.",
                path, pcap_snapshot(pcap));
#endif

    /* loop through the pcap.  get_next_packet() builds the cache for us! */
    while ((pktdata = get_next_packet(pcap, &pkthdr, file_idx, prev_packet)) != NULL) {
        packetnum++;
    }

    /* mark this file as cached */
    options.file_cache[file_idx].cached = TRUE;
    pcap_close(pcap);
}

/**
 * replay a pcap file out an interface
 */
void
replay_file(int file_idx)
{
    char *path = options.files[file_idx];
    pcap_t *pcap = NULL;
    char ebuf[PCAP_ERRBUF_SIZE];
    int dlt;

    if (! HAVE_OPT(QUIET))
        notice("processing file: %s", path);

    /* close stdin if reading from it (needed for some OS's) */
    if (strncmp(path, "-", 1) == 0)
        if (close(1) == -1)
            warnx("unable to close stdin: %s", strerror(errno));

    /* read from pcap file if we haven't cached things yet */
    if (! (options.enable_file_cache || options.preload_pcap)) {
        if ((pcap = pcap_open_offline(path, ebuf)) == NULL)
            errx(-1, "Error opening pcap file: %s", ebuf);
    } else {
        if (!options.file_cache[file_idx].cached)
            if ((pcap = pcap_open_offline(path, ebuf)) == NULL)
                errx(-1, "Error opening pcap file: %s", ebuf);

    }

#ifdef ENABLE_VERBOSE
    if (options.verbose) {

        /* in cache mode, we may not have opened the file */
        if (pcap == NULL)
            if ((pcap = pcap_open_offline(path, ebuf)) == NULL)
                errx(-1, "Error opening pcap file: %s", ebuf);

        /* init tcpdump */
        tcpdump_open(options.tcpdump, pcap);
    }
#endif


	if(1 == send_pkt_driver_mode){
	    if (pcap != NULL) {
	        dlt = sendpacket_get_dlt(options.intf1);
	        if ((dlt > 0) && (dlt != pcap_datalink(pcap)))
	            warnx("%s DLT (%s) does not match that of the outbound interface: %s (%s)", 
	                path, pcap_datalink_val_to_name(pcap_datalink(pcap)), 
	                options.intf1->device, pcap_datalink_val_to_name(dlt));
	    }
	}
    send_packets(pcap, file_idx);
    if (pcap != NULL)
        pcap_close(pcap);

#ifdef ENABLE_VERBOSE
    tcpdump_close(options.tcpdump);
#endif
}

/**
 * Initialize globals
 */
void
init(void)
{
    bytes_sent = failed = pkts_sent = 0;
    memset(&options, 0, sizeof(options));

    /* replay packets only once */
    options.loop = 1;

    /* Default mode is to replay pcap once in real-time */
    options.speed.mode = SPEED_MULTIPLIER;
    options.speed.speed = 1.0;

    /* Set the default timing method */
#ifdef HAVE_ABSOLUTE_TIME
    /* This is always the best (if the OS supports it) */
    options.accurate = ACCURATE_ABS_TIME;
#else
    /* This is probably the second best solution */
    options.accurate = ACCURATE_GTOD;
#endif

    /* set the default MTU size */
    options.mtu = DEFAULT_MTU;

    /* disable limit send */
    options.limit_send = -1;

#ifdef ENABLE_VERBOSE
    /* clear out tcpdump struct */
    options.tcpdump = (tcpdump_t *)safe_malloc(sizeof(tcpdump_t));
#endif

    cache_bit = cache_byte = 0;

#ifdef TCPBURST
	options.stream_multiple = 0;
	options.pkt_distance = 0;
	options.driver_mode = "pcap"; /* default mode is pcap */
	options.encap_cfg_file = NULL;
	char *env_var = getenv("MAX_BURST_PKTLEN");
	if(env_var != NULL)
	{
		options.max_burst_pkt_len = atoi(env_var);	
	}
	else
	{
		options.max_burst_pkt_len = TCP_BURST_MTU;
	}
	env_var = getenv("MIN_BURST_PKTLEN");
	if(env_var != NULL)
	{
		options.min_burst_pkt_len = atoi(env_var);	
	}
	else
	{
		options.min_burst_pkt_len = 0;
	}
#endif

		if (fcntl(STDERR_FILENO, F_SETFL, O_NONBLOCK) < 0)
			warnx("Unable to set STDERR to non-blocking: %s", strerror(errno));
}

/**
 * post processes the args and puts them into our options
 */
void
post_args(void)
{
    char *temp, *intname;
    char ebuf[SENDPACKET_ERRBUF_SIZE];
    int int1dlt, int2dlt;

#ifdef ENABLE_PCAP_FINDALLDEVS
    interface_list_t *intlist = get_interface_list();
#else
    interface_list_t *intlist = NULL;
#endif

#ifdef DEBUG
    if (HAVE_OPT(DBUG))
        debug = OPT_VALUE_DBUG;
#else
    if (HAVE_OPT(DBUG))
        warn("not configured with --enable-debug.  Debugging disabled.");
#endif

    options.loop = OPT_VALUE_LOOP;
    options.sleep_accel = OPT_VALUE_SLEEP_ACCEL;

#ifdef TCPBURST
#ifndef TCPREPLAY_EDIT
	options.stream_multiple = OPT_VALUE_MULTIPLE;
	if(options.stream_multiple < 0 || options.stream_multiple > 255)
	{
		printf("Invalid stream_multiple value:%d, must be[0-255].\n", options.stream_multiple);
		exit(1);
	}
	options.pkt_distance = OPT_VALUE_DISTANCE;
	if(options.pkt_distance < 0 || options.pkt_distance > 1024)
	{
		printf("Invalid pkt_distance value:%d, must be[0-1024].\n", options.pkt_distance);
		exit(2);
	}
	if((options.pkt_distance != 0) && (options.stream_multiple == 0))
	{
		printf("Warning, You assign pkt_distance but not stream_multiple, nothing to do!\n");
		sleep(1);
	}

	options.driver_mode = OPT_VALUE_DRIVER_MODE;
	if(NULL == options.driver_mode){
		options.driver_mode = "pcap"; /* 没指定参数, default is pcap */
	}	
	if((strncasecmp(options.driver_mode, "pcap", 4) != 0)
	&& (strncasecmp(options.driver_mode, "marsio", 6) != 0)){
		printf("Invalid driver_mode value:%s, must be[pcap,marsio].\n", options.driver_mode);
		exit(1);
	}

	options.encap_cfg_file = OPT_VALUE_ENCAP;
	if(options.encap_cfg_file != NULL){
		if(parse_encap_cfg_file(options.encap_cfg_file) < 0){
			printf("\n", options.encap_cfg_file);
			exit(1);
		}
	}
#endif	

#endif

    if (HAVE_OPT(LIMIT))
        options.limit_send = OPT_VALUE_LIMIT;

    if (HAVE_OPT(TOPSPEED)) {
        options.speed.mode = SPEED_TOPSPEED;
        options.speed.speed = 0.0;
    } else if (HAVE_OPT(PPS)) {
        options.speed.mode = SPEED_PACKETRATE;
        options.speed.speed = (float)OPT_VALUE_PPS;
        options.speed.pps_multi = OPT_VALUE_PPS_MULTI;
    } else if (HAVE_OPT(ONEATATIME)) {
        options.speed.mode = SPEED_ONEATATIME;
        options.speed.speed = 0.0;
    } else if (HAVE_OPT(MBPS)) {
        options.speed.mode = SPEED_MBPSRATE;
        options.speed.speed = atof(OPT_ARG(MBPS));
    } else if (HAVE_OPT(MULTIPLIER)) {
        options.speed.mode = SPEED_MULTIPLIER;
        options.speed.speed = atof(OPT_ARG(MULTIPLIER));
    }

    if (HAVE_OPT(STATS))
        options.stats = OPT_ARG(STATS);

#ifdef ENABLE_VERBOSE
    if (HAVE_OPT(VERBOSE))
        options.verbose = 1;

    if (HAVE_OPT(DECODE))
        options.tcpdump->args = safe_strdup(OPT_ARG(DECODE));

#endif

    /*
     * Check if the file cache should be enabled - if we're looping more than
     * once and the command line option has been spec'd
     */
    if (HAVE_OPT(ENABLE_FILE_CACHE) && (options.loop != 1)) {
        options.enable_file_cache = TRUE;
    }

    if (HAVE_OPT(PRELOAD_PCAP)) {
        options.preload_pcap = TRUE;
        options.enable_file_cache = TRUE;
    }

    if (HAVE_OPT(TIMER)) {
        if (strcmp(OPT_ARG(TIMER), "select") == 0) {
#ifdef HAVE_SELECT
            options.accurate = ACCURATE_SELECT;
#else
            err(-1, "tcpreplay not compiled with select support");
#endif
        } else if (strcmp(OPT_ARG(TIMER), "rdtsc") == 0) {
#ifdef HAVE_RDTSC
            options.accurate = ACCURATE_RDTSC;
#else
            err(-1, "tcpreplay not compiled with rdtsc support");
#endif
        } else if (strcmp(OPT_ARG(TIMER), "ioport") == 0) {
#if defined HAVE_IOPERM && defined(__i386__)
            options.accurate = ACCURATE_IOPORT;
            ioport_sleep_init();
#else
            err(-1, "tcpreplay not compiled with IO Port 0x80 support");
#endif
        } else if (strcmp(OPT_ARG(TIMER), "gtod") == 0) {
            options.accurate = ACCURATE_GTOD;
        } else if (strcmp(OPT_ARG(TIMER), "nano") == 0) {
            options.accurate = ACCURATE_NANOSLEEP;
        } else if (strcmp(OPT_ARG(TIMER), "abstime") == 0) {
#ifdef HAVE_ABSOLUTE_TIME
            options.accurate = ACCURATE_ABS_TIME;
            if  (!MPLibraryIsLoaded()) {
                err(-1, "The MP library did not load.\n");
            }
#else
            err(-1, "tcpreplay only supports absolute time on Apple OS X");
#endif
        } else {
            errx(-1, "Unsupported timer mode: %s", OPT_ARG(TIMER));
        }
    }

#ifdef HAVE_RDTSC
    if (HAVE_OPT(RDTSC_CLICKS)) {
        rdtsc_calibrate(OPT_VALUE_RDTSC_CLICKS);
    }
#endif

    if (HAVE_OPT(PKTLEN))
        warn("--pktlen may cause problems.  Use with caution.");

	if(1 == send_pkt_driver_mode){
	    if ((intname = get_interface(intlist, OPT_ARG(INTF1))) == NULL)
	        errx(-1, "Invalid interface name/alias: %s", OPT_ARG(INTF1));

	    options.intf1_name = safe_strdup(intname);

	    /* open interfaces for writing */
	    if ((options.intf1 = sendpacket_open(options.intf1_name, ebuf, TCPR_DIR_C2S)) == NULL)
	        errx(-1, "Can't open %s: %s", options.intf1_name, ebuf);

	    int1dlt = sendpacket_get_dlt(options.intf1);

	    if (HAVE_OPT(INTF2)) {
	        if ((intname = get_interface(intlist, OPT_ARG(INTF2))) == NULL)
	            errx(-1, "Invalid interface name/alias: %s", OPT_ARG(INTF2));

	        options.intf2_name = safe_strdup(intname);

	        /* open interface for writing */
	        if ((options.intf2 = sendpacket_open(options.intf2_name, ebuf, TCPR_DIR_S2C)) == NULL)
	            errx(-1, "Can't open %s: %s", options.intf2_name, ebuf);

	        int2dlt = sendpacket_get_dlt(options.intf2);
	        if (int2dlt != int1dlt)
	            errx(-1, "DLT type missmatch for %s (%s) and %s (%s)", 
	                    options.intf1_name, pcap_datalink_val_to_name(int1dlt), 
	                    options.intf2_name, pcap_datalink_val_to_name(int2dlt));
	    }

	    if (HAVE_OPT(CACHEFILE)) {
	        temp = safe_strdup(OPT_ARG(CACHEFILE));
	        options.cache_packets = read_cache(&options.cachedata, temp,
	                &options.comment);
	        safe_free(temp);
	    }
	}
    if (! HAVE_OPT(QUIET))
        notice("sending out %s %s", options.intf1_name,
                options.intf2_name == NULL ? "" : options.intf2_name);
}

/*
   Local Variables:
mode:c
indent-tabs-mode:nil
c-basic-offset:4
End:
*/