michaelh/zephyr
1
0
Fork 0
Code Releases Activity

Samples: Add SMP pktqueue application

Browse source 
This sample application simulates a network router with several
interfaces which performs IP Header Validation(calculation
of CRC16 header checksum). Each interface is represented
by sender queue(initialized with packet headers) and receiver queue.
Every header first is fetched from sender queue, than the header
CRC16 is calculated, and finally if hashsum is correct
the header is stored in receiver queue.
Each interface is can be processed independently by multiple threads.

This application can be used for testing the correctness
of synchronization mechanisms on multi-core systems.

Signed-off-by: Evgeniy Didin <didin@synopsys.com>
This commit is contained in:
Evgeniy Didin 2020-02-25 16:22:19 +03:00 • committed by Anas Nashif
commit 7371f97097
8 changed files with 422 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

10
samples/smp/pktqueue/CMakeLists.txt Normal file
View file

@ -0,0 +1,10 @@
# SPDX-License-Identifier: Apache-2.0
cmake_minimum_required(VERSION 3.13.1)
include($ENV{ZEPHYR_BASE}/cmake/app/boilerplate.cmake NO_POLICY_SCOPE)
project(smp_pktqueue)
FILE(GLOB pktqueue_sources src/*.c)
target_sources(app PRIVATE ${pktqueue_sources})

84
samples/smp/pktqueue/README.rst Normal file
View file

@ -0,0 +1,84 @@
.. _smp_pktqueue:
SMP pktqueue
############
Overview
********
This sample application performs a simplified network layer forwarding function
(essentially checksum calculation from IP Header Validation) of the Internet protocol
suite specified in RFC1812 "Requirements for IP Version 4 Routers" which
can be found at http://www.faqs.org/rfcs/rfc1812.html. This application
provides an indication of the potential performance of a microprocessor in an
IP router system.
At the beginning of the application the array (size defined in SIZE_OF_QUEUE)
of packet headers is initialized. Each header contains some random data of size
defined in SIZE_OF_HEADER and calculated crc16 header checksum
in appropriate field defined by CRC_BYTE_1 and CRC_BYTE_2. The contents of
header follows:
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| 0 - 3 | 4 - 7 | 8 - 15 | 16 - 31 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|Version| IHL |Type of Service| Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Identification |Flags| Fragment Offset |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Time to Live | Protocol | Header Checksum |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Source Address |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Address |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Options | Padding |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
The headers then are stored in multiple "sender" queues (the number is defined
in QUEUE_NUM). After that for each pair of "sender"/"receiver" queues one thread
is created, which will control "sender" queue processing.
Then in each queue-related thread several(defined in THREADS_NUM) threads are created. Each
of them first pick the header from "sender" queue, calculates crc and if
crc is correct put the header to "receiver" queue. Only one thread in a
time can access to sender or receiver queue.
As soon as all headers in each pair of queues are moved from "sender" to
"receiver" queue the execution of threads(related to pair) are terminated.
By changing the value of CONFIG_MP_NUM_CPUS on SMP systems, you
can see that using more cores takes almost linearly less time
to complete the computational task.
You can also edit the sample source code to change the
number of parallel executed pairs of queues(``QUEUE_NUM``),
the number of threads per pair of queues(``THREADS_NUM``),
the number of headers in queue (``SIZE_OF_QUEUE``), and
size of header in bytes (``SIZE_OF_HEADER``).
Building and Running
********************
This project outputs total time required for processing all packet headers.
It can be built and executed on QEMU as follows:
.. zephyr-app-commands::
:zephyr-app: samples/smp_pktqueue
:host-os: unix
:board: qemu_x86_64
:goals: run
:compact:
Sample Output
=============
.. code-block:: console
Simulating IP header validation on multiple cores.
Each of 2 parallel queues is processed by 3 threads on 2 cores and contain 5000 packet headers.
Bytes in packet header: 24
RESULT: OK
Application ran successfully.
All 20000 packet headers were processed in 89 msec

7
samples/smp/pktqueue/prj.conf Normal file
View file

@ -0,0 +1,7 @@
# Allow worker threads to capture all resources
CONFIG_MAIN_THREAD_PRIORITY=11
CONFIG_HEAP_MEM_POOL_SIZE=4096
CONFIG_TEST_RANDOM_GENERATOR=y
# Enable SMP
CONFIG_SMP=y

17
samples/smp/pktqueue/sample.yaml Normal file
View file

@ -0,0 +1,17 @@
sample:
description: Processing multiple queues in
a number of threads
name: SMP Pktqueue
common:
tags: introduction
harness: console
harness_config:
type: multi_line
ordered: yes
regex:
- "RESULT: OK(.*)"
tests:
sample.smp.pktqueue:
tags: introduction
filter: (CONFIG_MP_NUM_CPUS > 1)

199
samples/smp/pktqueue/src/main.c Normal file
View file

@ -0,0 +1,199 @@
/*
* Copyright (c) 2020 Synopsys, Inc.
*
* SPDX-License-Identifier: Apache-2.0
*/
#include "pktqueue.h"
#include "main.h"
static struct k_thread tthread[THREADS_NUM*QUEUE_NUM];
static struct k_thread qthread[QUEUE_NUM];
/* Each queue has its own mutex */
struct k_mutex sender_queue_mtx[QUEUE_NUM];
struct k_mutex receiver_queue_mtx[QUEUE_NUM];
/* Variable which indicates the amount of processed queues */
int queues_remain = QUEUE_NUM;
/* Variable to define current queue in thread */
int current_queue;
/* Array of packet header descriptors */
struct phdr_desc descriptors[QUEUE_NUM][SIZE_OF_QUEUE];
/* Arrays of receiver and sender queues */
struct phdr_desc_queue sender[QUEUE_NUM], receiver[QUEUE_NUM];
/* Array of packet headers */
uint8_t headers[QUEUE_NUM][SIZE_OF_QUEUE][SIZE_OF_HEADER];
static K_THREAD_STACK_ARRAY_DEFINE(tstack, THREADS_NUM*QUEUE_NUM, STACK_SIZE);
static K_THREAD_STACK_ARRAY_DEFINE(qstack, QUEUE_NUM, STACK_SIZE);
K_MUTEX_DEFINE(fetch_queue_mtx);
/* Function for initializing "sender" packet header queue */
void init_datagram_queue(struct phdr_desc_queue *queue, int queue_num)
{
queue->head = descriptors[queue_num];
for (int i = 0; i < SIZE_OF_QUEUE; i++) {
queue->tail = &descriptors[queue_num][i];
descriptors[queue_num][i].ptr = (uint8_t *)&headers[queue_num][i];
/* Fill packet header with random values */
for (int j = 0; j < SIZE_OF_HEADER; j++) {
/* leave crc field zeroed */
if (j < CRC_BYTE_1 || j > CRC_BYTE_2)
descriptors[queue_num][i].ptr[j] = (uint8_t)sys_rand32_get();
else
descriptors[queue_num][i].ptr[j] = 0;
}
/* Compute crc for further comparisson */
uint16_t crc;
crc = crc16(descriptors[queue_num][i].ptr, SIZE_OF_HEADER,
POLYNOMIAL, 0, 0);
/* Save crc value in header[CRC_BYTE_1-CRC_BYTE_2] field */
descriptors[queue_num][i].ptr[CRC_BYTE_1] = (uint8_t)(crc >> 8);
descriptors[queue_num][i].ptr[CRC_BYTE_2] = (uint8_t)(crc);
queue->count++;
descriptors[queue_num][i].next = &descriptors[queue_num][i+1];
}
}
/* Thread takes packet from "sender" queue and puts it to "receiver" queue.
* Each queue can be accessed only by one thread in a time. */
void test_thread(void *arg1, void *arg2, void *arg3)
{
struct phdr_desc_queue *sender_queue = (struct phdr_desc_queue *)arg1;
struct phdr_desc_queue *receiver_queue = (struct phdr_desc_queue *)arg2;
struct phdr_desc *qin_ptr = NULL;
int queue_num = *(int *)arg3;
/* Fetching one queue */
uint16_t crc, crc_orig;
qin_ptr = phdr_desc_dequeue(sender_queue, &sender_queue_mtx[queue_num]);
while (qin_ptr != NULL) {
/* Store original crc value from header */
crc_orig = qin_ptr->ptr[CRC_BYTE_1] << 8;
crc_orig |= qin_ptr->ptr[11];
/* Crc field should be zero before crc calculation */
qin_ptr->ptr[CRC_BYTE_1] = 0;
qin_ptr->ptr[CRC_BYTE_2] = 0;
crc = crc16(qin_ptr->ptr, SIZE_OF_HEADER, POLYNOMIAL, 0, 0);
/* Compare computed crc with crc from phdr_desc->crc */
if (crc == crc_orig) {
phdr_desc_enqueue(receiver_queue, qin_ptr,
&receiver_queue_mtx[queue_num]);
}
/* Take next element from "sender queue" */
qin_ptr = phdr_desc_dequeue(sender_queue,
&sender_queue_mtx[queue_num]);
}
}
/* Thread that processes one pair of sender/receiver queue */
void queue_thread(void *arg1, void *arg2, void *arg3)
{
ARG_UNUSED(arg1);
ARG_UNUSED(arg2);
ARG_UNUSED(arg3);
int queue_num;
/* Fetching one queue */
k_mutex_lock(&fetch_queue_mtx, K_FOREVER);
queue_num = current_queue;
current_queue++;
k_mutex_unlock(&fetch_queue_mtx);
for (int i = 0; i < THREADS_NUM; i++)
k_thread_create(&tthread[i+THREADS_NUM*queue_num],
tstack[i+THREADS_NUM*queue_num], STACK_SIZE,
(k_thread_entry_t)test_thread,
(void *)&sender[queue_num],
(void *)&receiver[queue_num], (void *)&queue_num,
K_PRIO_PREEMPT(10), 0, K_NO_WAIT);
/* Wait until sender queue is not empty */
while (sender[queue_num].count != 0)
k_sleep(K_MSEC(1));
/* Decrementing queue counter */
k_mutex_lock(&fetch_queue_mtx, K_FOREVER);
queues_remain--;
k_mutex_unlock(&fetch_queue_mtx);
}
void main(void)
{
uint32_t start_time, stop_time, cycles_spent, nanoseconds_spent;
current_queue = 0;
printk("Simulating IP header validation on multiple cores.\n");
printk("Each of %d parallel queues is processed by %d threads"
" on %d cores and contain %d packet headers.\n",
QUEUE_NUM, THREADS_NUM, CONFIG_MP_NUM_CPUS, SIZE_OF_QUEUE);
printk("Bytes in packet header: %d\n\n", SIZE_OF_HEADER);
/* initializing "sender" queue */
for (int i = 0; i < QUEUE_NUM; i++) {
init_datagram_queue(&sender[i], i);
k_mutex_init(&sender_queue_mtx[i]);
k_mutex_init(&receiver_queue_mtx[i]);
}
/* Capture initial time stamp */
start_time = k_cycle_get_32();
for (int i = 0; i < QUEUE_NUM; i++)
k_thread_create(&qthread[i], qstack[i], STACK_SIZE,
(k_thread_entry_t)queue_thread,
(void *)&sender[i], (void *)&receiver[i],
(void *)&i, K_PRIO_PREEMPT(11), 0, K_NO_WAIT);
/* Wait until all queues are not processed */
while (queues_remain > 0)
k_sleep(K_MSEC(1));
/* Capture final time stamp */
stop_time = k_cycle_get_32();
cycles_spent = stop_time - start_time;
nanoseconds_spent = (uint32_t)k_cyc_to_ns_floor64(cycles_spent);
/* Verify result of packet transmission
* The counter of correct receiver queues */
int correct = 0;
struct phdr_desc *tmp;
/* Iterate and count amount of packages in receiver queues */
for (int i = 0; i < QUEUE_NUM; i++) {
int count = 0;
tmp = receiver[i].head;
while (tmp != NULL) {
tmp = tmp->next;
count++;
}
if (receiver[i].count == SIZE_OF_QUEUE && count == SIZE_OF_QUEUE)
correct++;
}
if (correct == QUEUE_NUM)
printk("RESULT: OK\n"
"Application ran successfully.\n"
"All %d headers were processed in %d msec\n",
SIZE_OF_QUEUE*QUEUE_NUM,
nanoseconds_spent / 1000 / 1000);
else
printk("RESULT: FAIL\n"
"Application failed.\n"
"The amount of packets in receiver queue "
"is less than expected.\n");
k_sleep(K_MSEC(10));
}

32
samples/smp/pktqueue/src/main.h Normal file
View file

@ -0,0 +1,32 @@
/*
* Copyright (c) 2020 Synopsys, Inc.
*
* SPDX-License-Identifier: Apache-2.0
*/
#include <zephyr.h>
#include <stdio.h>
#include <sys/crc.h>
#include <random/rand32.h>
/* Amount of parallel processed sender/receiver queues of packet headers */
#define QUEUE_NUM 2
/* Amount of execution threads per pair of queues*/
#define THREADS_NUM (CONFIG_MP_NUM_CPUS+1)
/* Amount of packet headers in a queue */
#define SIZE_OF_QUEUE 5000
/* Size of packet header (in bytes) */
#define SIZE_OF_HEADER 24
/* CRC16 polynomial */
#define POLYNOMIAL 0x8005
/* CRC bytes in the packet */
#define CRC_BYTE_1 10
#define CRC_BYTE_2 11
#define STACK_SIZE 2048

46
samples/smp/pktqueue/src/pktqueue.c Normal file
View file

@ -0,0 +1,46 @@
/*
* Copyright (c) 2020 Synopsys, Inc.
*
* SPDX-License-Identifier: Apache-2.0
*/
#include "pktqueue.h"
/* Put a packet header in a queue defined in argument */
void phdr_desc_enqueue(struct phdr_desc_queue *queue, struct phdr_desc *desc,
struct k_mutex *mutex)
{
/* Locking queue */
k_mutex_lock(mutex, K_FOREVER);
if (queue->count == 0) {
queue->head = queue->tail = desc;
} else {
queue->tail->next = desc;
queue->tail = desc;
}
queue->count++;
desc->next = NULL;
/* Unlocking queue */
k_mutex_unlock(mutex);
}
/* Take a packet header from queue defined in argument */
struct phdr_desc *phdr_desc_dequeue(struct phdr_desc_queue *queue,
struct k_mutex *mutex)
{
struct phdr_desc *return_ptr = NULL;
/* Locking queue */
k_mutex_lock(mutex, K_FOREVER);
if (queue->count != 0) {
queue->count--;
return_ptr = queue->head;
queue->head = queue->head->next;
}
/* Unlocking queue */
k_mutex_unlock(mutex);
return return_ptr;
}

27
samples/smp/pktqueue/src/pktqueue.h Normal file
View file

@ -0,0 +1,27 @@
/*
* Copyright (c) 2020 Synopsys, Inc.
*
* SPDX-License-Identifier: Apache-2.0
*/
#include <zephyr.h>
#include <stdio.h>
#include <sys/crc.h>
#include <random/rand32.h>
struct phdr_desc {
struct phdr_desc *next; /* Next pkt descriptor in respective queue */
uint8_t *ptr; /* Pointer to header */
};
struct phdr_desc_queue {
struct phdr_desc *head; /* packet headers are removed from here */
struct phdr_desc *tail; /* packet headers are added here*/
int count;
};
void phdr_desc_enqueue(struct phdr_desc_queue *queue, struct phdr_desc *desc,
struct k_mutex *mutex);
struct phdr_desc *phdr_desc_dequeue(struct phdr_desc_queue *queue,
struct k_mutex *mutex);