字数统计: 3.4k字   |   阅读时长: 16分

NFD-C添加兴趣包优先级和推送机制

1. 添加兴趣包和数据包优先级

在之前添加了多个队列的基础上修改代码。

首先在tlv.h中添加新的兴趣包优先级的字段

1
TLV_InterestLevel = 47

然后就需要修改interest.h和interest.c中的各个函数。

首先需要为interest_t添加一个成员

1
uint8_t priority;
  • interest_tlv_size()函数
1
2
//新添加 priority
interest_buffer_size += 3;
  • interest_init()函数
1
2
//新添加初始化
interest->priority = 0;
  • interest_from_name()函数
1
2
//新添加
interest->priority = 0;
  • interest_tlv_encode()编码函数
1
2
3
4
5
6
7
8
9
10
//新添加priority, 表示兴趣包的优先级,priority和hop limit都是uint8_t类型
ret_val = encoder_append_type(enc, TLV_InterestLevel);
if (ret_val != 0)
  return ret_val;
ret_val = encoder_append_length(enc, 1);
if (ret_val != 0)
  return ret_val;
ret_val = encoder_append_byte(enc, interest->hop_limit);
if (ret_val != 0)
  return ret_val;
  • interest_from_tlv()解码函数
1
2
3
4
5
6
7
8
9
10
//新添加
else if (type == TLV_InterestLevel)
{
   ret_val = decoder_get_length(&dec, &length);
   if (ret_val != 0)
     return ret_val;
   ret_val = decoder_get_byte(&dec, &interest->hop_limit);
   if (ret_val != 0)
     return ret_val;
}
  • 新添加,设置优先级和获取优先级函数
1
2
3
4
5
6
7
8
9
void interest_set_priority(struct interest_t *interest, uint8_t priority)
{
 if (priority)
 interest->priority = priority;
}
uint8_t interest_get_priority(struct interest_t *interest)
{
 return interest->priority;
}

然后就是在forwarder中修改转发流程，根据interest包得到的priority判断发送到哪个队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
//在face.c中修改
void send_interest(struct face_t *face, struct interest_t *interest, Queue_Type* pInterestQueue, Queue_Type* pDataQueue, Queue_Type* pInterestQueue2, Queue_Type* pDataQueue2, Queue_Type* pInterestQueue3, Queue_Type* pDataQueue3, Queue_Type* pInterestQueue4, Queue_Type* pDataQueue4, Queue_Type* pInterestQueue5, Queue_Type* pDataQueue5, Queue_Type* pInterestQueue6, Queue_Type* pDataQueue6, Queue_Type* pInterestQueue7, Queue_Type* pDataQueue7, Queue_Type* pInterestQueue8, Queue_Type* pDataQueue8) {
  //修改，加入兴趣包的优先级
  uint8_t priority = interest_get_priority(interest);
  uint8_t buf[2048];
  struct encoder enc;
  encoder_init(&enc, buf, 2048);    //初始size是2048，offset是0
  int rtn = interest_tlv_encode(&enc, interest);
    if (rtn != 0) {
    ZF_LOGE("interest encode error");
    return;
  }
  //新增加的
  struct content_t content;
  content.face = face;
  content.enc = enc;
  //Queue_Type* pInterestQueue = InterestQueueInit();   //不应该在此初始化，应该考虑将队列作为参数传进来
  if(priority == 0)
  Queue_AddOne(pInterestQueue, &content);
  else if(priority == 1)
  Queue_AddOne(pInterestQueue2, &content);
  else if(priority == 2)
  Queue_AddOne(pInterestQueue3, &content);
  else if(priority == 3)
  Queue_AddOne(pInterestQueue4, &content);
  else if(priority == 4)
  Queue_AddOne(pInterestQueue5, &content);
  else if(priority == 5)
  Queue_AddOne(pInterestQueue6, &content);
  else if(priority == 6)
  Queue_AddOne(pInterestQueue7, &content);
  else
  Queue_AddOne(pInterestQueue8, &content);

  //queueInterestTest(pInterestQueue);	//进入检测兴趣队列函数, 应该考虑加在主函数中
  //入队，如果队列不为空，调用face.send
  //if(face->status)
  //  face->send(face, enc.buf, enc.offset);
}

//队列检测函数,优先级高的先发送
void queueInterestTest(Queue_Type* pInterestQueue, Queue_Type* pInterestQueue2, Queue_Type* pInterestQueue3, Queue_Type* pInterestQueue4, Queue_Type* pInterestQueue5, Queue_Type* pInterestQueue6, Queue_Type* pInterestQueue7, Queue_Type* pInterestQueue8) {  //主函数中调用完send后调用queueInterestTest真正发送
	while (pInterestQueue->Size != 0){
        struct content_t InterestValue;
        Queue_GetOne(pInterestQueue, &InterestValue); //GetOne的同时就出队了
        struct face_t* face;
        face = InterestValue.face;
        struct encoder enc;
        enc = InterestValue.enc;
        face->send(face, enc.buf, enc.offset);
    }
  while (pInterestQueue2->Size != 0){
        struct content_t InterestValue;
        Queue_GetOne(pInterestQueue2, &InterestValue); //GetOne的同时就出队了
        struct face_t* face;
        face = InterestValue.face;
        struct encoder enc;
        enc = InterestValue.enc;
        face->send(face, enc.buf, enc.offset);
    }
  ...
}

2. 在NFD-C中添加数据推送机制

数据推送就是生产者直接向消费者发送数据包数据，所有订阅了数据的消费者都可以接收数据。

修改流程：首先和兴趣包优先级修改类似，也需要修改数据包的字段，加入ifpush字段表示是普通数据包还是推送包。

首先在tlv.h中添加新类型

1
2
//新添加
TLV_Ifpush = 48

然后修改data.c 和 data.h

1
2
3
4
5
6
7
8
9
//data.h新添加ifpush和nonce字段
/**
 * The nonce of the Data.
 */
uint32_t nonce;
/**
 * If Data is PUSH data
 */
uint8_t ifpush;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
//data.c中修改编解码函数和计算size的函数
//_data_prepare_unsigned_block
//新添加ifpush和nonce
// ifpush
  ret_val = encoder_append_type(enc, TLV_Ifpush);
  if (ret_val != 0)
    return ret_val;
  ret_val = encoder_append_length(enc, 1);
  if (ret_val != 0)
    return ret_val;
  ret_val = encoder_append_byte(enc, data->ifpush);
  if (ret_val != 0)
    return ret_val;
  // 如果是推送机制，nonce不为0
  if(data->ifpush != 0 && data->nonce == 0) {
    data->nonce = (uint32_t)time_now_ms();
  }
  ret_val = encoder_append_type(enc, TLV_Nonce);
  if (ret_val != 0)
    return ret_val;
  ret_val = encoder_append_length(enc, 4);
  if (ret_val != 0)
    return ret_val;
  ret_val = encoder_append_uint32(enc, data->nonce);
  if (ret_val != 0)
    return ret_val;

//data_init初始化函数
  data->ifpush = 0;
  data->nonce = 0;

//data_tlv_encode函数中计算size时加入ifpush和nonce的计算以及上面的编码函数
  // 新添加
  // ifpush
  data_buffer_size += 3;
  // nonce
  data_buffer_size += 6;

//data_tlv_encode_with_digest_sign函数进行与data_tlv_encode函数相同的修改

//最后print_data函数增加打印输出
  printf("Ifpush: %u\n", data->ifpush);
  printf("Nonce: %lu\n", (unsigned long)data->nonce);

//这样就可以为data包加入ifpush和nonce字段了
  • data包的编解码改完之后，正式进行推送的修改，首先定义了push-server和push分别对应数据的生产者和消费者，推送的流程是消费者通过set_data_filter()设置数据包过滤器，相当于订阅了数据，生产者通过expressdata()函数发送数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
//push-server.cpp，在ping.cpp的基础上进行修改
#include <iostream>
#include <sstream>
#include <unordered_map>
#include <thread>
#include <chrono>
#include <mutex>

#include <signal.h> // signal

#include "cmdline.h"

extern "C" {
#include <interest.h>
#include <data.h>
#include <app-face.h>
#include <util/ctime.h>
}

//不知道为什么在interest.c中新添加的interest_set_priority和interest_get_priority在tools文件夹中读不到，这里面的interest.h一直是修改前的版本，编译后也不能用

void run(std::string prefix, int interval);
void on_data(const struct data_t *data, void *userdata);
//void on_timeout(const struct interest_t *interest, void *userdata);
void on_sigint(int sig);

struct app_face_t *app_face;
volatile bool push_running;
uint64_t send_count = 1, recv_count = 0, timeout_count = 0;
double max = 0, ave = 0, min = __DBL_MAX__;

std::mutex hash_mutex;
std::unordered_map<std::string, time_us_t> hash;

int main(int argc, char *argv[]) {
	cmdline::parser a;
  a.add<std::string>("prefix", 'p', "push prefix", true, "");
  a.add<int>("interval", 'i', "push interval", false, 1000);
	a.parse_check(argc, argv);
	std::string prefix = a.get<std::string>("prefix");
	int interval = a.get<int>("interval");

	if (::signal(SIGINT, on_sigint) == SIG_ERR)
  {
    std::cerr << "signal error" << std::endl;
  }

	app_face = app_face_init();
	push_running = true;
	std::thread t(run, prefix, interval); //子线程去执行run不断发包

	process_events(app_face); //主线程执行process_events监测app_face
	t.join();
  app_face_destory(app_face);

	return 0;
}

void run(std::string prefix, int interval) {    //主要对应发送兴趣包
	struct name_t data_name;
  struct data_t data;
  while (push_running)
  {
		std::stringstream ss;
		ss << prefix << '/' << send_count;  //相当于发送的是命令行中定义的prefix/send_count(代表发到第几个包了)
    name_from_string(&data_name, ss.str().c_str(), ss.str().size());  //此函数把string打包成interest_name，形成interest包的名字块
    data_from_name(&data, &data_name);  //生成一个新的interest包
    data.ifpush = 1;
    //如果考虑添加优先级应该在此添加，ping发送命令时就包括了兴趣包的优先级
    //uint8_t priority = 0;
    //interest_set_priority(&interest, priority);
		hash_mutex.lock();
    hash[ss.str()] = time_now_us(); //将名字ss.str()映射为现在的时间
		hash_mutex.unlock();

    express_data(app_face, &data, on_data, NULL); //把兴趣包发送出去，注意这里调用的是app_face中的express_interest
		std::cout << "send interest " << ss.str() << std::endl;

    memset(&data_name, 0, sizeof(struct name_t));
    memset(&data, 0, sizeof(struct data_t));

    std::this_thread::sleep_for(std::chrono::milliseconds(interval)); //睡眠iterval秒
    ++ send_count;
  }
}

void on_data(const struct data_t *data, void *userdata)
{
  time_us_t now = time_now_us();
  ++recv_count;

  char data_name_str[name_string_length(&data->name)];
  name_to_string(&data->name, data_name_str);
  std::string data_name(data_name_str, strlen(data_name_str));

	hash_mutex.lock();
  time_us_t sendTime = hash[data_name];
	hash_mutex.unlock();
  double delay = (now - sendTime) / (1000.0);
	printf("send data: %s -- %.2lfms\n", data_name.c_str(), delay);

  max = max > delay ? max : delay;
  min = min < delay ? min : delay;
  ave = (ave * (recv_count - 1) + delay) / recv_count;
}

void on_timeout(const struct interest_t *interest, void *userdata)
{
  // ++timeout_count;

  // char interest_name_str[name_string_length(&interest->name)];
  // name_to_string(&interest->name, interest_name_str);
  // printf("interest: %s time out\n", interest_name_str);
}

void on_sigint(int sig) //当触发终止信号时输出send的信息
{
	push_running = false;
  printf("send: %lu, recv: %lu, timeout: %lu\n", send_count, recv_count, timeout_count);
  printf("delay: max = %.2lfms, ave = %.2lfms, min = %.2lfms\n", max, ave, min);

  app_face_stop(app_face);
  exit(0);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
//push.cpp，在ping-server.cpp的基础上进行修改
#include <iostream>

#include "cmdline.h"

extern "C" {
#include <interest.h>
#include <data.h>
#include <app-face.h>
#include <util/ctime.h>
}

struct app_face_t *app_face;

void on_pushdata(const struct interest_t *interest, void *userdata);

int main(int argc, char *argv[]) {
	cmdline::parser a;
    a.add<std::string>("prefix", 'p', "push prefix", true, "");
	a.parse_check(argc, argv);
	std::string prefix = a.get<std::string>("prefix");
	std::cout << "push-receiver " << prefix << std::endl;

	app_face = app_face_init();

	set_data_filter(app_face, prefix.c_str(), on_pushdata, NULL);

	process_events(app_face);
  	app_face_destory(app_face);

	return 0;
}

void on_pushdata(const struct data_t *data, void *userdata) {
  	char data_name_str[name_string_length(&data->name)];
  	name_to_string(&data->name, data_name_str);
	printf("recv data: %s\n", data_name_str);

 	//struct data_t data;
 	//data_from_name(&data, &interest->name);
	//put(app_face, &data);
}

为了实现上面的推送机制，还需要定义set_data_filter()和express_data()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
//set_data_filter的实现
//app-face.h新增加
struct data_filter_t {
	struct name_t prefix;
	void (*on_pushdata)(const struct data_t *, void *);
	void *userdata;
};
struct app_face_t
{
	...
	//新增加,PDT暂时没有用
	ht_t *pending_data_table;
	struct data_filter_t * data_filter;
	...
};

//新增加，生产者端express_data，这里面的回调对应push-server中的on_data，用来统计推送信息
void express_data(struct app_face_t *app_face,
											struct data_t *data,
                      void (*on_data)(const struct data_t *, void *),
                      void *userdata)
{ //on_data和on_timeout在nfdc-status中，处理一些输出的状态
  if (app_face == NULL || data == NULL)
  {
    ZF_LOGW("app_face or interest is NULL in express_interest");
    return;
  }

  char data_name[name_string_length(&data->name)];
  name_to_string(&data->name, data_name);
  
  // printf("app-face interest flags: %lld\n",interest->flags);
	uint8_t buf[2048];
  memset(buf, 0, 2048);
  struct encoder enc;
  encoder_init(&enc, buf, 2048);
  data_tlv_encode_with_digest_sign(&enc, data);  //对数据包包进行TLV编码，从网络层到数据链路层

  app_face_send(app_face, enc.buf, enc.offset); //发送出去，实际就是发送到套接字文件中
}

//新增加set_data_filter，用来实现订阅，这里的回调对应push.cpp的on_pushdata，用来输出接收的数据包信息，订阅会向本地的NFD发送一个添加路由的兴趣包，添加此prefix对应的下一跳是本地的应用层face，这样当此节点的NFD收到对应前缀的包时，会直接发送给应用层
void set_data_filter(struct app_face_t *app_face,
													const char *prefix,
													void (*on_pushdata)(const struct data_t *, void *),
													void *userdata) 
{ //新增加，订阅机制，消费者向自己的NFD订阅消息,on_pushdata是对推送消息的回调
  if (prefix == NULL || on_pushdata == NULL)
  {
    ZF_LOGW("prefix or on_interest is NULL in set_interest_filter");
  }

  size_t prefix_len = strlen(prefix);
	name_from_string(&(app_face->data_filter->prefix), prefix, prefix_len);
	app_face->data_filter->on_pushdata = on_pushdata;
	app_face->data_filter->userdata = userdata;

	// Name: /localhost/route/add/prefix/app
	// /a/b/c -> a:b:c
  char prefix_copy[prefix_len + 1];
  strcpy(prefix_copy, prefix);
  for (int i = 0; i < prefix_len; ++ i) {
    if (prefix_copy[i] == '/') {
      prefix_copy[i] = ':';
    }
  }

	char str[50];
	sprintf(str, "/localhost/route/add/%s/app", prefix_copy);
	struct name_t interest_name;
  struct interest_t interest;
	name_from_string(&interest_name, str, strlen(str));
  interest_from_name(&interest, &interest_name);
  interest_set_canBePrefix(&interest, false);
  interest_set_mustBeFresh(&interest, true);
  // interest_set_traceroute(&interest, false);

  uint8_t buf[2048];
  memset(buf, 0, 2048);
  struct encoder enc;
  encoder_init(&enc, buf, 2048);
  interest_tlv_encode(&enc, &interest);

  app_face_send(app_face, enc.buf, enc.offset); //向本节点的NFD发送管理兴趣包，告知自身注册的哪些兴趣包前缀，可以提供服务
}

//还需要修改app-face的handle_msg，原来这里接收到数据包还需要与PIT进行对比，这里新增了处理推送包的逻辑，正常包走原来的通道，推送包直接调用data-filter的回调，当push接收到push-server的推送包后会进入此回调
    else if (type == TLV_Data)  //这部分只有消费者端进行处理
    {
      struct data_t *data = (struct data_t *)malloc(sizeof(struct data_t));
      data_init(data);
      if(data_tlv_decode_with_digest_verify(data, packet, packet_len) != 0)   //解码：将数据链路层的数据转换为网络层的数据
        break;

      char data_name[name_string_length(&data->name)];
      name_to_string(&data->name, data_name);
      
      if(data->ifpush == 0) {
      pthread_mutex_lock(&app_face->app_face_mutex);
      struct pending_interest_t *pi = (struct pending_interest_t *)(htget(app_face->pending_interest_table, data_name, strlen(data_name)));
      pthread_mutex_unlock(&(app_face->app_face_mutex));
      // printf("data before send to terminal: %s\n", data->value);

      if (pi != NULL) //如果PIT中有数据包条目，才会进行处理
      {
        cancel_timer(app_face->tq ,pi->expire_timer);
        if (pi->on_data)
          pi->on_data(data, pi->userdata);  //进入处理数据的管道
        pthread_mutex_lock(&(app_face->app_face_mutex));
        htrm(app_face->pending_interest_table, data_name, strlen(data_name)); //删除PIT表项
        pthread_mutex_unlock(&(app_face->app_face_mutex));

        free(pi->interest);
        free(pi);
      }
      }
      //处理推送包
      else {
        app_face->data_filter->on_pushdata(data, app_face->data_filter->userdata);  //调用data_filter对应的回调，具体实现在push-server中
      }

      free(data);
    }

此外，在转发层面上也需要进行修改，原来的数据包是根据PIT的in_record来实现反向发送的，而推送包没有PIT直接发送，现有的想法是手动建立路由，直接根据路由表转发。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
//首先修改on_incoming_data管道,如果是push数据，进入on_incoming_Pushdata管道
void on_incoming_data(struct data_t *data, struct face_t *face)
{
  struct name_t *name = &data->name;
  //新添加, 如果是push数据，进入on_incoming_Pushdata管道
  if(data->ifpush != 0) {
    on_incoming_Pushdata(data, face);
  }
  else {
      ...
  }
}

//on_incoming_Pushdata会缓存数据，然后根据进入on_outgoing_pushdata根据路由表转发
void on_incoming_Pushdata(struct data_t *data, struct face_t *face) 
{
  //如果此节点注册了前缀信息，就不会对数据进行转发，fib的下一跳是本节点的nfd，否则就会进行转发
  cs_insert(data);
  //对Pushdata根据路由表转发，进入on_outgoing_data管道
  on_outgoing_pushdata(data);
}

//on_outgoing_pushdata()函数：根据FIB进行数据包广播转发
static void on_outgoing_pushdata(struct data_t *data)
{
  fib_entry_t *fib_entry = fib_find_longest_prefix_matchs(&data->name); //根据FIB最长前缀匹配寻找下一条列表
    if (NULL == fib_entry)
  {
    //原来的部分
    ZF_LOGW("no route");
    return;
  }

  nexthop_t *nexthop_list = fib_entry->nexthop_list; //nexthop_list是一个链表
  if (nexthop_list == NULL)
  {
    //原来的部分
    ZF_LOGW("no route");
    return;
  }

  ZF_LOGI("fib_entry = %p, face id = %d", fib_entry, nexthop_list->face->id); //找到下一跳的faceid

  //muticast
  while (nexthop_list != NULL)  //对所有下一条列表广播,可能有本地节点，下一跳节点等等
  {

    if (nexthop_list->face->status == FACE_STATE_UP)
    {
      // printf("out face id is %d\n", nexthop_list->face->id);
      send_data(nexthop_list->face, data);  //根据FIB向下一条的face发送兴趣包,添加到队列里
      //queueDataTest(pInterestQueue, pDataQueue, pInterestQueue2, pDataQueue2, pInterestQueue3, pDataQueue3, pInterestQueue4, pDataQueue4, pInterestQueue5, pDataQueue5, pInterestQueue6, pDataQueue6, pInterestQueue7, pDataQueue7, pInterestQueue8, pDataQueue8);    //检测，实际上就是发送
      break;
    }
    nexthop_list = nexthop_list->next;
  }
}

一些编译方面的问题：

nfdc/tools用到了usr/local中的ndn库，这个库是nfdc目录下编译时生成的，因此如果想要改变兴趣包或数据包结构在app中用，不能直接make clean，make；需要make clean，make build，make install三步，具体每步的作用参考makefile。

然后makefile实际上是cmake根据cmakelist.txt生成的，因此如果想要在app中新添文件，就需要修改app目录下的cmakelist.txt，把新文件如何编译添加进txt文件中，这样编译时就会编译新添加的文件了。

ffmpeg -i input.flv -c copy -f mp4 output.mp4

扫一扫，分享到微信

微信分享二维码
本站总访问量次 |

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

这一定是一个<br><br>很长很长的一生啊