字数统计: 2.7k字   |   阅读时长: 14分

NFDC添加缓存队列

main函数的基本流程：每个节点创建一个socket套接字(server_sock)用作监听，epoll用于检测文件描述符。每当检测到监听描述符就会通过unix_stream_face_create新建一个face。当检测到其他的描述符时说明需要通信了。会通过刚刚创建的face的face_recv为本地节点接收数据，存到recv_msg中，recv_msg包括face，起始的buf地址，数据的大小。然后调用线程中的handle_msg对recv_msg进行处理，进一步判断收到的包的类型，如何转发等(后面的on_incoming_xxx)。

起始的interest是通过app_face的express_interest发出来的,比如ping工具。

添加队列的思路：

新建一个结构体content_t，content_t包括face和enc，content类型作为队列存储的类型，用于后面的发送。

1
2
3
4
5
struct content_t
{
  struct face_t *face;
  struct encoder enc;
};

添加一个新的.c文件，或者直接用标准库的queue或者deque

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
125
126
127
128
129
130
131
132
133
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <assert.h>

typedef content_t QDateType;

typedef struct QueueNode
{
	QDateType val;
	struct QueueNode* next;
}QueueNode;

typedef	struct Queue
{
	QueueNode* head;
	QueueNode* tail;
}Queue;

void QueueInit(Queue* pq)
{
	assert(pq);
	pq->head = pq->tail = NULL;
}

void QueueDestory(Queue* pq)
{
	assert(pq);
	QueueNode* cur = pq->head;
	while (cur)
	{
		QueueNode* next = cur->next;
		free(cur);
		cur = next;
	}
	pq->tail = pq->head = NULL;
}

void QueuePush(Queue* pq, QDateType x)
{
	assert(pq);

	QueueNode* newNode = (QueueNode*)malloc(sizeof(QueueNode));
	if (NULL == newNode)
	{
		printf("malloc error\n");
		exit(-1);
	}
	newNode->val = x;
	newNode->next = NULL;

	if (pq->tail == NULL)
	{
		assert(pq->head == NULL);
		pq->head = pq->tail = newNode;
	}
	else
	{
		pq->tail->next = newNode;
		pq->tail = newNode;
	}

}

void QueuePop(Queue* pq)
{
	assert(pq);
	assert(pq->head && pq->tail);
	if (pq->head->next == NULL)
	{
		free(pq->head);
		pq->head = pq->tail = NULL;
	}
	else
	{
		QueueNode* next = pq->head->next;
		free(pq->head);
		pq->head = next;
	}
}

bool QueueEmpty(Queue* pq)
{
	assert(pq);

	return pq->head == NULL;
}

QDateType QueueFront(Queue* pq)
{
	assert(pq);
	assert(pq->head);

	return pq->head->val;
}

int QueueSize(Queue* pq)
{
	assert(pq);
	QueueNode* cur = pq->head;
	int count = 0;
	while (cur)
	{
		cur = cur->next;
		count++;
	}
	return count;
}

//测试
int main()
{
	Queue Q;
	QueueInit(&Q);
	QueuePush(&Q, 1);
	QueuePush(&Q, 2);
	QueuePush(&Q, 3);
	QueuePush(&Q, 4);
	QueuePush(&Q, 5);
	QDateType ret1 = QueueFront(&Q);
	printf("QueueFront:%d\n", ret1);
	//QDateType ret2 = QueueBack(&Q);
	//printf("QueueBack:%d\n", ret2);
	int size = QueueSize(&Q);
	printf("size：%d\n", size);
	while (!QueueEmpty(&Q))
	{
		printf("%d", QueueFront(&Q));
		QueuePop(&Q);
	}
	QueueDestory(&Q);
	return 0;
}

对send_interest进行修改原来是直接通过接口发送，现在是将数据发送到队列中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//原来的版本
void send_interest(struct face_t *face, struct interest_t *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;
  }
  //入队，如果队列不为空，调用face.send
  if(face->status)
    face->send(face, enc.buf, enc.offset);
}

系统调用write的作用是把缓冲区buf里的前nbytes字节写入与文件描述符相关联的文件中去。它的返回值是实际写出的字节数。

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
//新的版本
//定义两个全局变量queueInterest和queueData
Queue queueInterest;
Queue queueData;
QueueInit(&queueInterest);
QueueInit(&queueData);
void send_interest(struct face_t *face, struct interest_t *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 = (struct content_t *)malloc(sizeof(struct content_t));
  content->face = face;
  content->enc = enc;
  //RQueue<content_t> queueInterest;
  QueuePush(&queueInterest, content);
  queueInterestTest(queueInterest);	//进入检测兴趣队列函数
  //入队，如果队列不为空，调用face.send
  //if(face->status)
  //  face->send(face, enc.buf, enc.offset);
}
1
2
3
4
5
6
7
8
void queueInterestTest(RQueue<content_t> queueInterest) {
	while(!queueInterest.isempty()){
        face_t face = QueueFront(&queueInterest).face;
        encoder enc = QueueFront(&queueInterest).enc;
        QueuePop(&queueInterest);
        face->send(face, enc.buf, enc.offset);
    }
}
1
2
3
4
5
6
7
8
void queueDataTest(RQueue<content_t> queueInterest) {
	while(queueInterest.isempty() && !queueData.isempty()){
        face_t face = QueueFront(&queueData).face;
        encoder enc = QueueFront(&queueData).enc;
        QueuePop(&queueData);
        face->send(face, enc.buf, enc.offset);
    }
}

具体实现：

  1. 在main函数中新建两个队列并修改接口
1
2
3
4
5
6
//新增
Queue_Type* pQueue0 = InterestQueueInit();
Queue_Type* pDataQueue = DataQueueInit();
//修改
on_incoming_interest(interest, recv_msg->face, pInterestQueue, pDataQueue);     //修改过！
on_incoming_data(data, recv_msg->face, pInterestQueue, pDataQueue);
  1. 修改forwarder和mgmt里面对应的接口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
//mgmt
void on_mgmt_interest(struct name_t *name, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
void on_mib_interest(struct name_t *name, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
void on_cs_interest(struct name_t *name, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);

//forwarder
void on_incoming_interest(struct interest_t *interest, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
void on_incoming_data(struct data_t *data, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
static void on_contentstore_hit(pit_entry_t *pit_entry, struct data_t *data, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue); 
static void on_contentstore_miss(pit_entry_t *pit_entry, struct interest_t *interest, struct face_t *face, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
static void on_outgoing_interest(struct interest_t *interest, pit_entry_t *pit_entry, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);
static void on_outgoing_data(pit_entry_t *pit_entry, struct data_t *data, Queue_Type* pInterestQueue, Queue_Type* pDataQueue);

//forwarder中具体发送时的接口，并添加queueInterestTest和queueDataTest，send只是将数据发送到队列中，Test中负责具体发送
send_data_to_queue(face, data, pInterestQueue, pDataQueue);
queueDataTest(pQueue0, pQueue1);
send_interest_to_queue(nexthop_list->face, interest, Queue0, Queue1); 
queueInterestTest(pQueue0, pQueue1);    //检测，实际上就是发送
  1. 对face进行修改
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
//send_interest函数
void send_interest(struct face_t *face, struct interest_t *interest, Queue_Type* pInterestQueue, Queue_Type* pDataQueue) {
  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();   //不应该在此初始化，应该考虑将队列作为参数传进来
  Queue_AddOne(pInterestQueue, &content);
  //queueInterestTest(pInterestQueue);	//进入检测兴趣队列函数, 应该考虑加在主函数中
  //入队，如果队列不为空，调用face.send
  //if(face->status)
  //  face->send(face, enc.buf, enc.offset);
}

void queueInterestTest(Queue_Type* pInterestQueue) {  //主函数中调用完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);
    }
}

void queueDataTest(Queue_Type* pDataQueue, Queue_Type* pInterestQueue) {
	while (pDataQueue->Size != 0 && pInterestQueue == 0) {
        struct content_t DataValue;
        Queue_GetOne(pDataQueue, &DataValue); //GetOne的同时就出队了
        struct face_t* face;
        face = DataValue.face;
        struct encoder enc;
        enc = DataValue.enc;
        face->send(face, enc.buf, enc.offset);
    }
}

队列的具体实现：

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
#ifndef __QUEUE_H_
#define __QUEUE_H_

/*********************************************************************
 * @brief       Queue队列类型定义
 *********************************************************************/
//queue.h
typedef struct
{
    unsigned int            Capacity;          // Queue容量 
    volatile unsigned int   Head;           // Head为起始元素
    volatile unsigned int   Tail;           // Tail-1为最后一个元素
    volatile unsigned int   Size;        // 元素个数
    unsigned int            ElementBytes;   // 每个元素的字节数
    void                    *Buff;          // 缓存区
}Queue_Type;
  
/*********************************************************************
 * @brief       Queue初始化
 * @param[in]   pQueue: Queue指针
 * @param[in]   pBuff: Queue中缓存
 * @param[in]   elementBytes:Queue每个元素的字节数
 * @param[in]   Capacity: Queue容量 
 * @return      None
 *********************************************************************/
void Queue_Init(Queue_Type *pQueue, void *pBuff, unsigned int elementBytes, unsigned int Capacity);
  
/*********************************************************************
 * @brief       向Queue添加一个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValue: 要添加的元素
 * @return      1-TRUE or 0-FALSE
 *********************************************************************/
unsigned char Queue_AddOne(Queue_Type *pQueue, void *pValue);
  
/*********************************************************************
 * @brief       向Queue添加多个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValues: 要添加的元素指针
 * @param[in]   bytesToAdd: 要添加元素的长度
 * @return      实际添加的元素个数
 *********************************************************************/
unsigned int Queue_Add(Queue_Type *pQueue, void *pValues, unsigned int bytesToAdd);
  
/*********************************************************************
 * @brief       从Queue读取一个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValue: 存放要读取的元素指针
 * @return      1-TRUE or 0-FALSE
 *********************************************************************/
unsigned char Queue_GetOne(Queue_Type *pQueue, void *pValue);
  
/*********************************************************************
 * @brief       从Queue读取多个元素
 * @param[in]   pQueue: Queue指针
 * @param[out]  pValues: 存放要读取的元素指针
 * @param[in]   bytesToRead: 要读取的元素长度
 * @return      实际读取的元素个数
 *********************************************************************/
unsigned int Queue_Get(Queue_Type *pQueue, void *pValues, unsigned int bytesToRead);
  
  
/*********************************************************************
 * @brief       清空Queue
 * @param[in]   pQueue: Queue指针
 * @return      None
 *********************************************************************/
void Queue_Clear(Queue_Type *pQueue);
  
#endif

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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
//queue.c
#include <string.h>
#include "queue.h"
  
/*********************************************************************
 * @brief       Queue初始化
 * @param[in]   pQueue: Queue指针
 * @param[in]   pBuff: Queue中缓存
 * @param[in]   elementBytes:Queue每个元素的字节数
 * @param[in]   depth: Queue深度
 * @return      None
 *********************************************************************/
void Queue_Init(Queue_Type *pQueue, void *pBuff, unsigned int elementBytes, unsigned int capacity)
{
    pQueue->Buff = pBuff;
    pQueue->ElementBytes = elementBytes;
    pQueue->Capacity = capacity;
    pQueue->Head = 0;
    pQueue->Tail = 0;
    pQueue->Size = 0;
}
  
/*********************************************************************
 * @brief       判断Queue是否为空
 * @param[in]   pQueue: Queue指针
 * @return      1-TRUE or 0-FALSE
 *********************************************************************/
unsigned char Queue_IsEmpty(Queue_Type *pQueue)
{
    return (pQueue->Size == 0);
}
  
/*********************************************************************
 * @brief       判断Queue是否已满
 * @param[in]   pQueue: Queue指针
 * @return      TRUE or FALSE
 *********************************************************************/
unsigned char Queue_IsFull(Queue_Type *pQueue)
{
    return (pQueue->Size == pQueue->Capacity);
}
  
/*********************************************************************
 * @brief       向Queue添加一个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValue: 要添加的元素
 * @return      1-TRUE or 0-FALSE
 *********************************************************************/
unsigned char Queue_AddOne(Queue_Type *pQueue, void *pValue)
{
    unsigned char *p;
  
    if (Queue_IsFull(pQueue))
    {
        return 0;
    }
  
    p = (unsigned char *)pQueue->Buff;
    memcpy(p + pQueue->Tail, (unsigned char *)pValue, pQueue->ElementBytes);
     
    pQueue->Tail += pQueue->ElementBytes;
    if (pQueue->Tail >= pQueue->Capacity)
    {
        pQueue->Tail = 0;
    }
    pQueue->Size ++;
    return 1;
}
  
/*********************************************************************
 * @brief       向Queue添加多个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValues: 要添加的元素指针
 * @param[in]   bytesToAdd: 要添加元素的长度
 * @return      实际添加的元素个数
 *********************************************************************/
unsigned int Queue_Add(Queue_Type *pQueue, void *pValues, unsigned int bytesToAdd)
{
    unsigned char *p;
    unsigned int cnt = 0;
  
    p = (unsigned char *)pValues;
    while(bytesToAdd --)
    {
        if (Queue_AddOne(pQueue, p))
        {
            p += pQueue->ElementBytes;
            cnt++;
        }
        else
        {
            break;
        }
    }
  
    return cnt;
}
  
/*********************************************************************
 * @brief       从Queue读取一个元素
 * @param[in]   pQueue: Queue指针
 * @param[in]   pValue: 存放要读取的元素指针
 * @return      1-TRUE or 0-FALSE
 *********************************************************************/
unsigned char Queue_GetOne(Queue_Type *pQueue, void *pValue)
{
    unsigned char *p;
    if (Queue_IsEmpty(pQueue))
    {
        return 0;
    }
  
    p = (unsigned char *)pQueue->Buff;
    memcpy(pValue, p + pQueue->Head, pQueue->ElementBytes);
  
    pQueue->Head += pQueue->ElementBytes;
    if (pQueue->Head >= pQueue->Capacity)
    {
        pQueue->Head = 0;
    }
    pQueue->Size --;
  
    return 1;
}
  
/*********************************************************************
 * @brief       从Queue读取多个元素
 * @param[in]   pQueue: Queue指针
 * @param[out]  pValues: 存放要读取的元素指针
 * @param[in]   bytesToRead: 要读取的元素长度
 * @return      实际读取的元素个数
 *********************************************************************/
unsigned int Queue_Get(Queue_Type *pQueue, void *pValues, unsigned int bytesToRead)
{
    unsigned int cnt = 0;
    unsigned char *p;
  
    p = pValues;
    while(bytesToRead--)
    {
        if (Queue_GetOne(pQueue, p))
        {
            p += pQueue->ElementBytes;
            cnt++;
        }
        else
        {
            break;
        }
    }
  
    return cnt;
}
  
/*********************************************************************
 * @brief       清空Queue
 * @param[in]   pQueue: Queue指针
 * @return      None
 *********************************************************************/
void Queue_Clear(Queue_Type *pQueue)
{
    pQueue->Size = 0;
    pQueue->Head = 0;
    pQueue->Tail = 0;
}

/*
void TestQueue(void)
{
    int i;
    Queue_Type   queue;
    Queue_Type   *pQueue;
    int         index;
    float       fArray[10];
    float       fValue;
  
    char        cArray[10];
    char        cValue;
  
    pQueue = &queue;
  
    printf("测试Queue元素为float型的数据\r\n");
    printf("初始化Queue值。\r\n");
    Queue_Init(pQueue, fArray, sizeof(float), 10);
    for (i = 0; i < 10; i++)
    {
        fValue = (100.0f+i*i);
        Queue_AddOne(pQueue, &fValue);
    }
    printf("当前元数个数：%d\r\n", pQueue->Size);
    if(Queue_IsFull(pQueue))
    printf("Queue Is Full\n");
    
    
    index = 0;
    while(Queue_GetOne(pQueue, &fValue))
    {
        index ++;
        printf("第%d个元素fValue = %0.3f\r\n",index, fValue);
        if (index == 5)
        {
            printf("插入3个值。\r\n");
            fValue = 1.23f;
            Queue_AddOne(pQueue, &fValue);
  
            fValue = 2.34f;
            Queue_AddOne(pQueue, &fValue);
  
            fValue = 3.45f;
            Queue_AddOne(pQueue, &fValue);
        }
    }
  
    printf("\r\n\r\n");
    printf("测试Queue元素为char型的数据\r\n");
    Queue_Init(pQueue, cArray, sizeof(char), 10);
    printf("初始化Queue值。\r\n");
    Queue_Add(pQueue, "ABCDEFGHIJ", 10);
    printf("当前元数个数：%d\r\n", pQueue->Size);
    index = 0;
    while(Queue_GetOne(pQueue, &cValue))
    {
        index ++;
        printf("第%d个元素cValue = %c\r\n",index, cValue);
        if (index == 2)		//如果超过队列长度，这个元素就不会插进去 
        {
            printf("插入3个值。\r\n");
            cValue = 'X';
            Queue_AddOne(pQueue, &cValue);
  
            cValue = 'Y';
            Queue_AddOne(pQueue, &cValue);
  
            cValue = 'Z';
            Queue_AddOne(pQueue, &cValue);
        }
    }
    
    struct content {
    	int age;
    	int gender;
	}; 
	
	struct content conArray[10];
	struct content conValue;
	printf("\r\n\r\n");
    printf("测试Queue元素为content型的数据\r\n");
    Queue_Init(pQueue, conArray, sizeof(conValue), 10);
    
    printf("初始化Queue值。\r\n");
    conValue.age = 100;
    conValue.gender = 1;
    Queue_AddOne(pQueue, &conValue);
    printf("当前元数个数：%d\r\n", pQueue->Size);
    
    index = 0;
	while(Queue_GetOne(pQueue, &conValue))
    {
        index ++;
        printf("第%d个元素conValue.age = %d, %d",index, conValue.age, conValue.gender);
    }
	
	
}
*/

扫一扫，分享到微信

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

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>很长很长的一生啊