maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate PNF socket send functions into separate socket library

Browse Source
This commit is contained in:
Rúben Soares Silva
2025-06-27 09:56:46 +01:00
parent f986f6843a
commit bfaec58ca5
18 changed files with 1089 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nfapi/open-nFAPI/pnf/inc/pnf.h
+1 -2
View File
@@ -38,9 +38,8 @@ struct pnf_t {
int pnf_connect(pnf_t *pnf);
int pnf_message_pump(pnf_t *pnf);
int pnf_nr_message_pump(pnf_t *pnf);
bool pnf_nr_send_p5_message(pnf_t* pnf, nfapi_nr_p4_p5_message_header_t* msg, uint32_t msg_len);
void pnf_nr_handle_p5_message(pnf_t* pnf, void* pRecvMsg, int recvMsgLen);
int pnf_pack_and_send_p5_message(pnf_t* pnf, nfapi_p4_p5_message_header_t* msg, uint32_t msg_len);
int pnf_pack_and_send_p4_message(pnf_t* pnf, nfapi_p4_p5_message_header_t* msg, uint32_t msg_len);
int pnf_send_message(pnf_t* pnf, uint8_t* msg, uint32_t msg_len, uint16_t stream_id);
nfapi/open-nFAPI/pnf/inc/pnf_p7.h
+6 -2
View File
@@ -144,7 +144,6 @@ struct pnf_p7_t {
int pnf_p7_message_pump(pnf_p7_t* pnf_p7);
int pnf_nr_p7_message_pump(pnf_p7_t* pnf_p7);
int pnf_p7_pack_and_send_p7_message(pnf_p7_t* pnf_p7, nfapi_p7_message_header_t* msg, uint32_t msg_len);
bool pnf_nr_send_p7_message(pnf_p7_t* pnf_p7, nfapi_nr_p7_message_header_t* header, uint32_t msg_len);
int pnf_p7_send_message(pnf_p7_t* pnf_p7, uint8_t* msg, uint32_t msg_len);
@@ -159,6 +158,11 @@ int nfapi_pnf_p7_nr_rach_ind(nfapi_pnf_p7_config_t* config, nfapi_nr_rach_indica
pnf_p7_rx_message_t* pnf_p7_rx_reassembly_queue_add_segment(pnf_p7_t* pnf_p7, pnf_p7_rx_reassembly_queue_t* queue, uint32_t rx_hr_time, uint16_t sequence_number, uint16_t segment_number, uint8_t m, uint8_t* data, uint16_t data_len);
void pnf_p7_rx_reassembly_queue_remove_msg(pnf_p7_t* pnf_p7, pnf_p7_rx_reassembly_queue_t* queue, pnf_p7_rx_message_t* msg);
void pnf_p7_rx_reassembly_queue_remove_old_msgs(pnf_p7_t* pnf_p7, pnf_p7_rx_reassembly_queue_t* queue, uint32_t rx_hr_time, uint32_t delta);
void pnf_nr_handle_p7_message(void *pRecvMsg, int recvMsgLen, pnf_p7_t *pnf_p7, uint32_t rx_hr_time);
struct timespec pnf_timespec_sub(struct timespec lhs, struct timespec rhs);
uint32_t pnf_get_current_time_hr(void);
struct timespec pnf_timespec_add(struct timespec lhs, struct timespec rhs);
void pnf_p7_free(pnf_p7_t* pnf_p7, void* ptr);
void* pnf_p7_malloc(pnf_p7_t* pnf_p7, size_t size);
#endif /* _PNF_P7_H_ */
nfapi/open-nFAPI/pnf/public_inc/nfapi_pnf_interface.h
+1 -2
View File
@@ -419,7 +419,6 @@ int nfapi_pnf_config_destroy(nfapi_pnf_config_t* config);
* \endcode
*/
int nfapi_pnf_start(nfapi_pnf_config_t* config);
int nfapi_nr_pnf_start(nfapi_pnf_config_t* config);
/*! Stop the PNF library.
* \param config A pointer to the pnf configuration
@@ -847,7 +846,7 @@ int nfapi_pnf_p7_config_destroy(nfapi_pnf_p7_config_t* config);
* This function will not return until nfapi_pnf_p7_stop is called.
*/
int nfapi_pnf_p7_start(nfapi_pnf_p7_config_t* config);
int nfapi_nr_pnf_p7_start(nfapi_pnf_p7_config_t* config);
/*! Stop the PNF P7 library.
* \param config A pointer to a PNF P7 config
nfapi/open-nFAPI/pnf/src/pnf.c
+25 -238
View File
@@ -30,6 +30,10 @@
#include "nfapi_nr_interface_scf.h"
#include "pnf.h"
#include <socket/include/socket_common.h>
#include "nfapi/open-nFAPI/fapi/inc/nr_fapi.h"
# if 1 // for hard-code (remove later)
#include "common/platform_types.h"
#include "common/platform_constants.h"
@@ -1705,20 +1709,6 @@ void pnf_handle_p5_message(pnf_t* pnf, void *pRecvMsg, int recvMsgLen)
}
}
bool pnf_nr_send_p5_message(pnf_t* pnf, nfapi_nr_p4_p5_message_header_t* msg, uint32_t msg_len)
{
int packed_len =
pnf->_public.pack_func(msg, msg_len, pnf->tx_message_buffer, sizeof(pnf->tx_message_buffer), &pnf->_public.codec_config);
if (packed_len < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "nfapi_nr_p5_message_pack failed (%d)\n", packed_len);
return false;
}
return pnf_send_message(pnf, pnf->tx_message_buffer, packed_len, 0 /*msg->stream_id*/) == 0;
}
int pnf_pack_and_send_p5_message(pnf_t* pnf, nfapi_p4_p5_message_header_t* msg, uint32_t msg_len)
{
int packed_len = nfapi_p5_message_pack(msg, msg_len,
@@ -1751,8 +1741,6 @@ int pnf_pack_and_send_p4_message(pnf_t* pnf, nfapi_p4_p5_message_header_t* msg,
return pnf_send_message(pnf, pnf->tx_message_buffer, packed_len, 0/*msg->stream_id*/);
}
int pnf_connect(pnf_t* pnf)
{
struct sockaddr_in servaddr;
@@ -1957,30 +1945,30 @@ int pnf_connect(pnf_t* pnf)
int pnf_send_message(pnf_t* pnf, uint8_t *msg, uint32_t len, uint16_t stream)
{
if (pnf->sctp)
{
if (pnf->sctp)
{
#if 0
printf("\nPNF SENDS: \n");
for(int i=0; i<len; i++){
printf("%d", msg[i]);
}
printf("\n");
printf("\nPNF SENDS: \n");
for(int i=0; i<len; i++){
printf("%d", msg[i]);
}
printf("\n");
#endif
if (sctp_sendmsg(pnf->p5_sock, msg, len, NULL, 0, 42/*config->sctp_stream_number*/, 0, stream/*P5_STREAM_ID*/, 0, 0) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "sctp_sendmsg failed errno: %d\n", errno);
return -1;
}
}
else
{
if (write(pnf->p5_sock, msg, len) != len)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "read failed errno: %d\n", errno);
return -1;
}
}
if (sctp_sendmsg(pnf->p5_sock, msg, len, NULL, 0, 42/*config->sctp_stream_number*/, 0, stream/*P5_STREAM_ID*/, 0, 0) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "sctp_sendmsg failed errno: %d\n", errno);
return -1;
}
}
else
{
if (write(pnf->p5_sock, msg, len) != len)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "read failed errno: %d\n", errno);
return -1;
}
}
return 0;
}
@@ -2117,146 +2105,6 @@ int pnf_read_dispatch_message(pnf_t* pnf)
}
int pnf_nr_read_dispatch_message(pnf_t* pnf)
{
int socket_connected = 1;
// 1. Peek the message header
// 2. If the message is larger than the stack buffer then create a dynamic buffer
// 3. Read the buffer
// 4. Handle the p5 message
uint32_t header_buffer_size = NFAPI_NR_P5_HEADER_LENGTH;
uint8_t header_buffer[header_buffer_size];
uint32_t stack_buffer_size = 32; // should it be the size of then sctp_notificatoin structure
uint8_t stack_buffer[stack_buffer_size];
uint8_t* dynamic_buffer = 0;
uint8_t* read_buffer = &stack_buffer[0];
uint32_t message_size = 0;
struct sockaddr_in addr;
socklen_t addr_len = sizeof(addr);
struct sctp_sndrcvinfo sndrcvinfo;
(void)memset(&sndrcvinfo, 0, sizeof(struct sctp_sndrcvinfo));
{
int flags = MSG_PEEK;
if (pnf->sctp) {
message_size = sctp_recvmsg(pnf->p5_sock,
header_buffer,
header_buffer_size,
/*(struct sockaddr*)&addr, &addr_len*/
0,
0,
&sndrcvinfo,
&flags);
} else {
message_size = recv(pnf->p5_sock,
header_buffer,
header_buffer_size,
flags);
}
if (message_size < NFAPI_NR_P5_HEADER_LENGTH) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF Failed to peek sctp message size errno:%d\n", errno);
return 0;
}
nfapi_nr_p4_p5_message_header_t header;
const bool result = pnf->_public.hdr_unpack_func(header_buffer, header_buffer_size, &header, sizeof(header), 0);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF Failed to unpack p5 message header\n");
return 0;
}
message_size = header.message_length + header_buffer_size;
// now have the size of the mesage
}
if (message_size > stack_buffer_size) {
dynamic_buffer = (uint8_t*)malloc(message_size);
if (dynamic_buffer == NULL) {
// todo : add error mesage
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF Failed to allocate dynamic buffer for sctp_recvmsg size:%d\n", message_size);
return -1;
}
read_buffer = dynamic_buffer;
}
{
int flags = 0;
(void)memset(&sndrcvinfo, 0, sizeof(struct sctp_sndrcvinfo));
ssize_t recvmsg_result = 0;
if (pnf->sctp) {
recvmsg_result = sctp_recvmsg(pnf->p5_sock,
read_buffer,
message_size,
(struct sockaddr *)&addr,
&addr_len,
&sndrcvinfo,
&flags);
} else {
if ((recvmsg_result = recv(pnf->p5_sock, read_buffer, message_size, flags))<=0) {
recvmsg_result = -1;
}
}
if (recvmsg_result == -1) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "Failed to read sctp message size errno:%d\n", errno);
} else {
#if 0
// print the received message
printf("\n MESSAGE RECEIVED: \n");
for(int i=0; i<message_size; i++){
printf("read_buffer[%d] = 0x%02x\n",i, read_buffer[i]);
}
printf("\n");
#endif
if (flags & MSG_NOTIFICATION) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "Notification received from %s:%u\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
// todo - handle the events
} else {
/*
NFAPI_TRACE(NFAPI_TRACE_INFO, "Received message fd:%d from %s:%u assoc:%d on stream %d, PPID %d, length %d, flags 0x%x\n",
pnf->p5_sock,
inet_ntoa(addr.sin_addr),
ntohs(addr.sin_port),
sndrcvinfo.sinfo_assoc_id,
sndrcvinfo.sinfo_stream,
ntohl(sndrcvinfo.sinfo_ppid),
message_size,
flags);
*/
// handle now if complete message in one or more segments
if ((flags & 0x80) == 0x80 || !pnf->sctp) {
pnf_nr_handle_p5_message(pnf, read_buffer, message_size);
} else {
NFAPI_TRACE(NFAPI_TRACE_WARN, "sctp_recvmsg: unhandled mode with flags 0x%x\n", flags);
// assume socket disconnected
NFAPI_TRACE(NFAPI_TRACE_WARN, "Disconnected socket\n");
socket_connected = 0;
}
}
}
}
if (dynamic_buffer) {
free(dynamic_buffer);
}
return socket_connected;
}
int pnf_message_pump(pnf_t* pnf)
{
uint8_t socketConnected = 1;
@@ -2308,67 +2156,6 @@ int pnf_message_pump(pnf_t* pnf)
pnf->_public.state = NFAPI_PNF_IDLE;
// close the connection and socket
if (close(pnf->p5_sock) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "close(sctpSock) failed errno: %d\n", errno);
}
return 0;
}
int pnf_nr_message_pump(pnf_t* pnf)
{
uint8_t socketConnected = 1;
while(socketConnected && pnf->terminate == 0)
{
fd_set rfds;
int selectRetval = 0;
// select on a timeout and then get the message
FD_ZERO(&rfds);
FD_SET(pnf->p5_sock, &rfds);
struct timeval timeout;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
selectRetval = select(pnf->p5_sock+1, &rfds, NULL, NULL, &timeout);
if(selectRetval == 0)
{
// timeout
continue;
}
else if (selectRetval == -1 && (errno == EINTR))
{
// interrupted by signal
NFAPI_TRACE(NFAPI_TRACE_WARN, "P5 Signal Interrupt %d\n", errno);
continue;
}
else if (selectRetval == -1)
{
NFAPI_TRACE(NFAPI_TRACE_WARN, "P5 select() failed\n");
sleep(1);
continue;
}
if(FD_ISSET(pnf->p5_sock, &rfds))
{
socketConnected = pnf_nr_read_dispatch_message(pnf);
}
else
{
NFAPI_TRACE(NFAPI_TRACE_WARN, "Why are we here\n");
}
}
// Drop back to idle if we have lost connection
pnf->_public.state = NFAPI_PNF_IDLE;
// close the connection and socket
if (close(pnf->p5_sock) < 0)
{
nfapi/open-nFAPI/pnf/src/pnf_interface.c
-30
View File
@@ -79,36 +79,6 @@ int nfapi_pnf_start(nfapi_pnf_config_t* config)
return 0;
}
int nfapi_nr_pnf_start(nfapi_pnf_config_t* config)
{
// Verify that config is not null
if(config == 0)
return -1;
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s\n", __FUNCTION__);
pnf_t* _this = (pnf_t*)(config);
while (_this->terminate == 0)
{
int connect_result = pnf_connect(_this);
if(connect_result > 0)
{
pnf_nr_message_pump(_this);
}
else if(connect_result < 0)
{
return connect_result;
}
sleep(1);
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() terminate=1 - EXITTING............\n", __FUNCTION__);
return 0;
}
int nfapi_pnf_stop(nfapi_pnf_config_t* config)
{
// Verify that config is not null
nfapi/open-nFAPI/pnf/src/pnf_p7.c
+9 -439
View File
@@ -32,7 +32,6 @@
#include "common/ran_context.h"
#include <SCHED_NR/phy_frame_config_nr.h>
#define FAPI2_IP_DSCP 0
extern int sf_ahead;
@@ -464,10 +463,10 @@ int pnf_p7_send_message(pnf_p7_t* pnf_p7, uint8_t* msg, uint32_t len)
socklen_t remote_addr_len = sizeof(struct sockaddr_in);
int sendto_result;
if ((sendto_result = sendto((int)pnf_p7->p7_sock, (const char*)msg, len, 0, (const struct sockaddr*)&remote_addr, remote_addr_len)) < 0)
//if ((sendto_result = sendto((int)pnf_p7->p7_sock,"hello", 6, 0, (const struct sockaddr*)&remote_addr, remote_addr_len)) < 0)
{
if ((sendto_result = sendto((int)pnf_p7->p7_sock, (const char*)msg, len, 0, (const struct sockaddr*)&remote_addr, remote_addr_len)) < 0)
//if ((sendto_result = sendto((int)pnf_p7->p7_sock,"hello", 6, 0, (const struct sockaddr*)&remote_addr, remote_addr_len)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "%s %s:%d sendto(%d, %p, %d) %d failed errno: %d\n", __FUNCTION__, pnf_p7->_public.remote_p7_addr, pnf_p7->_public.remote_p7_port, (int)pnf_p7->p7_sock, (const char*)msg, len, remote_addr_len, errno);
return -1;
}
@@ -569,87 +568,6 @@ int pnf_p7_pack_and_send_p7_message(pnf_p7_t* pnf_p7, nfapi_p7_message_header_t*
return 0;
}
bool pnf_nr_send_p7_message(pnf_p7_t* pnf_p7, nfapi_nr_p7_message_header_t* header, uint32_t msg_len)
{
header->m_segment_sequence = NFAPI_P7_SET_MSS(0, 0, pnf_p7->sequence_number);
// Need to guard against different threads calling the encode function at the same time
if (pthread_mutex_lock(&(pnf_p7->pack_mutex)) != 0) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "failed to lock mutex\n");
return false;
}
uint8_t tx_buf[131072]; // four times NFAPI_MAX_PACKED_MESSAGE_SIZE as of this commit
int len = pnf_p7->_public.pack_func(header, tx_buf, sizeof(tx_buf), &pnf_p7->_public.codec_config);
if (len < 0) {
if (pthread_mutex_unlock(&(pnf_p7->pack_mutex)) != 0) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "failed to unlock mutex\n");
return false;
}
NFAPI_TRACE(NFAPI_TRACE_ERROR, "nfapi_p7_message_pack failed with return %d\n", len);
return false;
}
if (len > pnf_p7->_public.segment_size) {
int msg_body_len = len - NFAPI_NR_P7_HEADER_LENGTH;
int seg_body_len = pnf_p7->_public.segment_size - NFAPI_NR_P7_HEADER_LENGTH;
int segment_count = (msg_body_len / (seg_body_len)) + ((msg_body_len % seg_body_len) ? 1 : 0);
int segment = 0;
int offset = NFAPI_NR_P7_HEADER_LENGTH;
uint8_t buffer[pnf_p7->_public.segment_size];
for (segment = 0; segment < segment_count; ++segment) {
uint8_t last = 0;
uint32_t size = pnf_p7->_public.segment_size - NFAPI_NR_P7_HEADER_LENGTH;
if (segment + 1 == segment_count) {
last = 1;
size = (msg_body_len) - (seg_body_len * segment);
}
uint32_t segment_size = size + NFAPI_NR_P7_HEADER_LENGTH;
// Update the header with the m and segement
memcpy(&buffer[0], tx_buf, NFAPI_NR_P7_HEADER_LENGTH);
// set the segment length
buffer[4] = (segment_size & 0xFF000000) >> 24;
buffer[5] = (segment_size & 0xFF0000) >> 16;
buffer[6] = (segment_size & 0xFF00) >> 8;
buffer[7] = (segment_size & 0xFF);
// set the m & segment number
buffer[8] = ((!last) << 7) + segment;
memcpy(&buffer[NFAPI_NR_P7_HEADER_LENGTH], tx_buf + offset, size);
offset += size;
if (pnf_p7->_public.checksum_enabled) {
nfapi_nr_p7_update_checksum(buffer, segment_size);
}
pnf_p7_send_message(pnf_p7, &buffer[0], segment_size);
}
} else {
if (pnf_p7->_public.checksum_enabled) {
nfapi_nr_p7_update_checksum(tx_buf, len);
}
// simple case that the message fits in a single segment
pnf_p7_send_message(pnf_p7, tx_buf, len);
}
pnf_p7->sequence_number++;
if (pthread_mutex_unlock(&(pnf_p7->pack_mutex)) != 0) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "failed to unlock mutex\n");
return false;
}
return true;
}
void pnf_pack_and_send_timing_info(pnf_p7_t* pnf_p7)
{
nfapi_timing_info_t timing_info;
@@ -2171,7 +2089,7 @@ void pnf_dispatch_p7_message(void *pRecvMsg, int recvMsgLen, pnf_p7_t* pnf_p7,
}
}
void pnf_nr_dispatch_p7_message(void* pRecvMsg, int recvMsgLen, pnf_p7_t* pnf_p7, uint32_t rx_hr_time)
void pnf_nr_handle_p7_message(void* pRecvMsg, int recvMsgLen, pnf_p7_t* pnf_p7, uint32_t rx_hr_time)
{
nfapi_nr_p7_message_header_t header;
@@ -2326,110 +2244,6 @@ void pnf_handle_p7_message(void *pRecvMsg, int recvMsgLen, pnf_p7_t* pnf_p7, ui
}
void pnf_nr_handle_p7_message(void* pRecvMsg, int recvMsgLen, pnf_p7_t* pnf_p7, uint32_t rx_hr_time)
{
nfapi_nr_p7_message_header_t messageHeader;
// validate the input params
if (pRecvMsg == NULL || recvMsgLen < 4 || pnf_p7 == NULL) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "pnf_handle_p7_message: invalid input params (%p %d %p)\n", pRecvMsg, recvMsgLen, pnf_p7);
return;
}
// unpack the message header
const bool result =
pnf_p7->_public.hdr_unpack_func(pRecvMsg, recvMsgLen, &messageHeader, sizeof(messageHeader), &pnf_p7->_public.codec_config);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Unpack message header failed, ignoring\n");
return;
}
uint8_t m = NFAPI_P7_GET_MORE(messageHeader.m_segment_sequence);
uint8_t sequence_num = NFAPI_P7_GET_SEQUENCE(messageHeader.m_segment_sequence);
uint8_t segment_num = NFAPI_P7_GET_SEGMENT(messageHeader.m_segment_sequence);
if (pnf_p7->_public.checksum_enabled) {
uint32_t checksum = nfapi_nr_p7_calculate_checksum(pRecvMsg, recvMsgLen);
if (checksum != messageHeader.checksum) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Checksum verification failed %d %d\n", checksum, messageHeader.checksum);
return;
}
}
if (m == 0 && segment_num == 0) {
// we have a complete message
// ensure the message is sensible
if (recvMsgLen < 8 || pRecvMsg == NULL) {
NFAPI_TRACE(NFAPI_TRACE_WARN, "Invalid message size: %d, ignoring\n", recvMsgLen);
return;
}
pnf_nr_dispatch_p7_message(pRecvMsg, recvMsgLen, pnf_p7, rx_hr_time);
} else {
pnf_p7_rx_message_t* rx_msg = pnf_p7_rx_reassembly_queue_add_segment(pnf_p7,
&(pnf_p7->reassembly_queue),
rx_hr_time,
sequence_num,
segment_num,
m,
pRecvMsg,
recvMsgLen);
if (rx_msg->num_segments_received == rx_msg->num_segments_expected) {
// send the buffer on
uint16_t i = 0;
uint32_t length = 0;
for (i = 0; i < rx_msg->num_segments_expected; ++i) {
length += rx_msg->segments[i].length - (i > 0 ? NFAPI_NR_P7_HEADER_LENGTH : 0);
}
if (pnf_p7->reassemby_buffer_size < length) {
pnf_p7_free(pnf_p7, pnf_p7->reassemby_buffer);
pnf_p7->reassemby_buffer = 0;
}
if (pnf_p7->reassemby_buffer == 0) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Resizing PNF_P7 Reassembly buffer %d->%d\n", pnf_p7->reassemby_buffer_size, length);
pnf_p7->reassemby_buffer = (uint8_t*)pnf_p7_malloc(pnf_p7, length);
if (pnf_p7->reassemby_buffer == 0) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Failed to allocate PNF_P7 reassemby buffer len:%d\n", length);
return;
}
memset(pnf_p7->reassemby_buffer, 0, length);
pnf_p7->reassemby_buffer_size = length;
}
uint32_t offset = 0;
for (i = 0; i < rx_msg->num_segments_expected; ++i) {
if (i == 0) {
memcpy(pnf_p7->reassemby_buffer, rx_msg->segments[i].buffer, rx_msg->segments[i].length);
offset += rx_msg->segments[i].length;
} else {
memcpy(pnf_p7->reassemby_buffer + offset,
rx_msg->segments[i].buffer + NFAPI_NR_P7_HEADER_LENGTH,
rx_msg->segments[i].length - NFAPI_NR_P7_HEADER_LENGTH);
offset += rx_msg->segments[i].length - NFAPI_NR_P7_HEADER_LENGTH;
}
}
pnf_nr_dispatch_p7_message(pnf_p7->reassemby_buffer, length, pnf_p7, rx_msg->rx_hr_time);
// delete the structure
pnf_p7_rx_reassembly_queue_remove_msg(pnf_p7, &(pnf_p7->reassembly_queue), rx_msg);
}
}
// The timeout used to be 1000, i.e., 1ms, which is too short. The below 10ms
// is selected to be able to encompass any "reasonable" slot ahead time for the VNF.
// Ideally, we would remove old msg (segments) if we detect packet loss
// (e.g., if the sequence numbers advances sufficiently); in the branch of
// this commit, our goal is to make the PNF work, so we content ourselves to
// just remove very old messages.
pnf_p7_rx_reassembly_queue_remove_old_msgs(pnf_p7, &(pnf_p7->reassembly_queue), rx_hr_time, 10000);
}
void pnf_nfapi_p7_read_dispatch_message(pnf_p7_t* pnf_p7, uint32_t now_hr_time)
{
int recvfrom_result = 0;
@@ -2499,68 +2313,6 @@ void pnf_nfapi_p7_read_dispatch_message(pnf_p7_t* pnf_p7, uint32_t now_hr_time)
while(recvfrom_result > 0);
}
void pnf_nr_nfapi_p7_read_dispatch_message(pnf_p7_t* pnf_p7, uint32_t now_hr_time)
{
int recvfrom_result = 0;
struct sockaddr_in remote_addr;
socklen_t remote_addr_size = sizeof(remote_addr);
remote_addr.sin_family = 2; // hardcoded
do {
// peek the header
uint8_t header_buffer[NFAPI_NR_P7_HEADER_LENGTH];
recvfrom_result = recvfrom(pnf_p7->p7_sock,
header_buffer,
NFAPI_NR_P7_HEADER_LENGTH,
MSG_DONTWAIT | MSG_PEEK,
(struct sockaddr*)&remote_addr,
&remote_addr_size);
if (recvfrom_result > 0) {
// get the segment size
nfapi_nr_p7_message_header_t header;
pnf_p7->_public.hdr_unpack_func(header_buffer, NFAPI_NR_P7_HEADER_LENGTH, &header, 34, 0);
// resize the buffer if we have a large segment
if (header.message_length > pnf_p7->rx_message_buffer_size) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "reallocing rx buffer %d\n", header.message_length);
pnf_p7->rx_message_buffer = realloc(pnf_p7->rx_message_buffer, header.message_length);
pnf_p7->rx_message_buffer_size = header.message_length;
}
// read the segment
recvfrom_result = recvfrom(pnf_p7->p7_sock,
pnf_p7->rx_message_buffer,
header.message_length,
MSG_DONTWAIT,
(struct sockaddr*)&remote_addr,
&remote_addr_size);
now_hr_time = pnf_get_current_time_hr(); // moved to here - get closer timestamp???
if (recvfrom_result > 0) {
pnf_nr_handle_p7_message(pnf_p7->rx_message_buffer, recvfrom_result, pnf_p7, now_hr_time);
// printf("\npnf_handle_p7_message sfn=%d,slot=%d\n",pnf_p7->sfn,pnf_p7->slot);
}
} else if (recvfrom_result == 0) {
// recv zero length message
recvfrom_result =
recvfrom(pnf_p7->p7_sock, header_buffer, 0, MSG_DONTWAIT, (struct sockaddr*)&remote_addr, &remote_addr_size);
}
if (recvfrom_result == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// return to the select
// NFAPI_TRACE(NFAPI_TRACE_WARN, "%s recvfrom would block :%d\n", __FUNCTION__, errno);
} else {
NFAPI_TRACE(NFAPI_TRACE_WARN, "%s recvfrom failed errno:%d\n", __FUNCTION__, errno);
}
}
// need to update the time as we would only use the value from the
// select
} while (recvfrom_result > 0);
}
int pnf_p7_message_pump(pnf_p7_t* pnf_p7)
{
@@ -2602,7 +2354,7 @@ int pnf_p7_message_pump(pnf_p7_t* pnf_p7)
}
*/
int iptos_value = FAPI2_IP_DSCP << 2;
int iptos_value = 0;
if (setsockopt(pnf_p7->p7_sock, IPPROTO_IP, IP_TOS, &iptos_value, sizeof(iptos_value)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF P7 setsockopt (IPPROTO_IP, IP_TOS) failed errno: %d\n", errno);
@@ -2720,197 +2472,15 @@ struct timespec pnf_timespec_add(struct timespec lhs, struct timespec rhs)
struct timespec pnf_timespec_sub(struct timespec lhs, struct timespec rhs)
{
struct timespec result;
if ((lhs.tv_nsec-rhs.tv_nsec)<0)
if ((lhs.tv_nsec-rhs.tv_nsec)<0)
{
result.tv_sec = lhs.tv_sec-rhs.tv_sec-1;
result.tv_nsec = 1000000000+lhs.tv_nsec-rhs.tv_nsec;
}
else
}
else
{
result.tv_sec = lhs.tv_sec-rhs.tv_sec;
result.tv_nsec = lhs.tv_nsec-rhs.tv_nsec;
}
return result;
}
int pnf_nr_p7_message_pump(pnf_p7_t* pnf_p7)
{
// initialize the mutex lock
if(pthread_mutex_init(&(pnf_p7->mutex), NULL) != 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P7 mutex init: %d\n", errno);
return -1;
}
if(pthread_mutex_init(&(pnf_p7->pack_mutex), NULL) != 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P7 mutex init: %d\n", errno);
return -1;
}
// create the pnf p7 socket
if ((pnf_p7->p7_sock = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P7 socket errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF P7 socket created (%d)...\n", pnf_p7->p7_sock);
// configure the UDP socket options
int reuseaddr_enable = 1;
if (setsockopt(pnf_p7->p7_sock, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_enable, sizeof(int)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF P7 setsockopt (SOL_SOCKET, SO_REUSEADDR) failed errno: %d\n", errno);
return -1;
}
/*
int reuseport_enable = 1;
if (setsockopt(pnf_p7->p7_sock, SOL_SOCKET, SO_REUSEPORT, &reuseport_enable, sizeof(int)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF P7 setsockopt (SOL_SOCKET, SO_REUSEPORT) failed errno: %d\n", errno);
return -1;
}
*/
int iptos_value = FAPI2_IP_DSCP << 2;
if (setsockopt(pnf_p7->p7_sock, IPPROTO_IP, IP_TOS, &iptos_value, sizeof(iptos_value)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF P7 setsockopt (IPPROTO_IP, IP_TOS) failed errno: %d\n", errno);
return -1;
}
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(pnf_p7->_public.local_p7_port);
if(pnf_p7->_public.local_p7_addr == 0)
{
addr.sin_addr.s_addr = INADDR_ANY;
}
else
{
//addr.sin_addr.s_addr = inet_addr(pnf_p7->_public.local_p7_addr);
if(inet_aton(pnf_p7->_public.local_p7_addr, &addr.sin_addr) == -1)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "inet_aton failed\n");
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF P7 binding %d too %s:%d\n", pnf_p7->p7_sock, inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
if (bind(pnf_p7->p7_sock, (struct sockaddr *)&addr, sizeof(addr)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF_P7 bind error fd:%d errno: %d\n", pnf_p7->p7_sock, errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF P7 bind succeeded...\n");
//Initializaing timing structures needed for slot ticking
struct timespec slot_start;
clock_gettime(CLOCK_MONOTONIC, &slot_start);
struct timespec pselect_start;
struct timespec slot_duration;
slot_duration.tv_sec = 0;
slot_duration.tv_nsec = 0.5e6;
//Infinite loop
while(pnf_p7->terminate == 0)
{
fd_set rfds;
int selectRetval = 0;
// select on a timeout and then get the message
FD_ZERO(&rfds);
FD_SET(pnf_p7->p7_sock, &rfds);
struct timespec timeout;
timeout.tv_sec = 100;
timeout.tv_nsec = 0;
clock_gettime(CLOCK_MONOTONIC, &pselect_start);
//setting the timeout
if((pselect_start.tv_sec > slot_start.tv_sec) || ((pselect_start.tv_sec == slot_start.tv_sec) && (pselect_start.tv_nsec > slot_start.tv_nsec)))
{
// overran the end of the subframe we do not want to wait
timeout.tv_sec = 0;
timeout.tv_nsec = 0;
//struct timespec overrun = pnf_timespec_sub(pselect_start, sf_start);
//NFAPI_TRACE(NFAPI_TRACE_INFO, "Subframe overrun detected of %d.%d running to catchup\n", overrun.tv_sec, overrun.tv_nsec);
}
else
{
// still time before the end of the subframe wait
timeout = pnf_timespec_sub(slot_start, pselect_start);
}
selectRetval = pselect(pnf_p7->p7_sock+1, &rfds, NULL, NULL, &timeout, NULL);
uint32_t now_hr_time = pnf_get_current_time_hr();
if(selectRetval == 0)
{
// timeout
//update slot start timing
slot_start = pnf_timespec_add(slot_start, slot_duration);
//increment sfn/slot
if (++pnf_p7->slot == 20)
{
pnf_p7->slot = 0;
pnf_p7->sfn = (pnf_p7->sfn + 1) % 1024;
}
continue;
}
else if (selectRetval == -1 && (errno == EINTR))
{
// interrupted by signal
NFAPI_TRACE(NFAPI_TRACE_WARN, "PNF P7 Signal Interrupt %d\n", errno);
continue;
}
else if (selectRetval == -1)
{
NFAPI_TRACE(NFAPI_TRACE_WARN, "PNF P7 select() failed\n");
sleep(1);
continue;
}
if(FD_ISSET(pnf_p7->p7_sock, &rfds))
{
pnf_nr_nfapi_p7_read_dispatch_message(pnf_p7, now_hr_time);
}
}
NFAPI_TRACE(NFAPI_TRACE_ERROR, "PNF_P7 Terminating..\n");
// close the connection and socket
if (close(pnf_p7->p7_sock) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "close failed errno: %d\n", errno);
}
if(pthread_mutex_destroy(&(pnf_p7->pack_mutex)) != 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "mutex destroy failed errno: %d\n", errno);
}
if(pthread_mutex_destroy(&(pnf_p7->mutex)) != 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "mutex destroy failed errno: %d\n", errno);
}
return 0;
}
nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c
-13
View File
@@ -74,20 +74,7 @@ int nfapi_pnf_p7_start(nfapi_pnf_p7_config_t* config)
return 0;
}
int nfapi_nr_pnf_p7_start(nfapi_pnf_p7_config_t* config)
{
// Verify that config is not null
if(config == 0)
return -1;
pnf_p7_t* _this = (pnf_p7_t*)(config);
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s\n", __FUNCTION__);
pnf_nr_p7_message_pump(_this);
return 0;
}
int nfapi_pnf_p7_stop(nfapi_pnf_p7_config_t* config)