yigid/openairinterface5g
1
0
Fork 0
mirror of https://gitlab.eurecom.fr/oai/openairinterface5g.git synced 2026-07-13 04:30:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Separate VNF socket send functions into separate socket library

Browse Source
This commit is contained in:
Rúben Soares Silva
2025-03-17 14:38:13 +00:00
parent bfaec58ca5
commit 9b71f1d324
12 changed files with 946 additions and 1037 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
nfapi/oai_integration/nfapi_vnf.c
View File

@@ -36,6 +36,13 @@
#include "nfapi_vnf.h"
#include <vnf.h>
#include <vnf_p7.h>
#ifdef ENABLE_SOCKET
#include <socket/include/socket_vnf.h>
#endif
#include <vnf.h>
#include <vnf_p7.h>
#include "nfapi.h"
#include "vendor_ext.h"
@@ -1391,16 +1398,9 @@ int vnf_nr_pack_p4_p5_vendor_extension(void *header, uint8_t **ppWritePackedMsg,
return 0;
}
static pthread_t vnf_start_pthread;
static pthread_t vnf_p5_init_and_receive_pthread;
static pthread_t vnf_p7_start_pthread;
void *vnf_nr_p7_start_thread(void *ptr) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
pthread_setname_np(pthread_self(), "VNF_P7");
nfapi_vnf_p7_config_t *config = (nfapi_vnf_p7_config_t *)ptr;
nfapi_nr_vnf_p7_start(config);
return config;
}
void *vnf_p7_start_thread(void *ptr) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
@@ -1410,7 +1410,7 @@ void *vnf_p7_start_thread(void *ptr) {
return config;
}
void *vnf_nr_p7_thread_start(void *ptr)
void *configure_nr_p7_vnf(void *ptr)
{
init_queue(&gnb_rach_ind_queue);
init_queue(&gnb_rx_ind_queue);
@@ -1450,9 +1450,15 @@ void *vnf_nr_p7_thread_start(void *ptr)
p7_vnf->config->unpack_func = &nfapi_nr_p7_message_unpack;
p7_vnf->config->hdr_unpack_func = &nfapi_nr_p7_message_header_unpack;
p7_vnf->config->pack_func = &nfapi_nr_p7_message_pack;
#ifdef ENABLE_SOCKET
p7_vnf->config->unpack_func = &nfapi_nr_p7_message_unpack;
p7_vnf->config->hdr_unpack_func = &nfapi_nr_p7_message_header_unpack;
p7_vnf->config->pack_func = &nfapi_nr_p7_message_pack;
p7_vnf->config->send_p7_msg = &vnf_nr_send_p7_msg;
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Creating VNF NFAPI P7 start thread %s\n", __FUNCTION__);
pthread_create(&vnf_p7_start_pthread, NULL, &vnf_nr_p7_start_thread, p7_vnf->config);
threadCreate(&vnf_p7_start_pthread, &vnf_nr_start_p7_thread, p7_vnf->config, "vnf_p7_thread", -1, OAI_PRIORITY_RT);
#endif
return 0;
}
@@ -1500,7 +1506,7 @@ int pnf_nr_start_resp_cb(nfapi_vnf_config_t *config, int p5_idx, nfapi_nr_pnf_st
if(p7_vnf->thread_started == 0) {
pthread_t vnf_p7_thread;
threadCreate(&vnf_p7_thread, &vnf_nr_p7_thread_start, p7_vnf, "vnf_p7_thread", -1, OAI_PRIORITY_RT);
threadCreate(&vnf_p7_thread, &configure_nr_p7_vnf, p7_vnf, "vnf_p7_thread", -1, OAI_PRIORITY_RT);
p7_vnf->thread_started = 1;
} else {
// P7 thread already running.
@@ -1510,7 +1516,10 @@ int pnf_nr_start_resp_cb(nfapi_vnf_config_t *config, int p5_idx, nfapi_nr_pnf_st
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Sending NFAPI_VNF_PARAM_REQUEST phy_id:%d\n", pnf->phys[0].id);
memset(&req, 0, sizeof(req));
req.header.message_id = NFAPI_NR_PHY_MSG_TYPE_PARAM_REQUEST;
#ifdef ENABLE_SOCKET
req.header.phy_id = pnf->phys[0].id;
#endif
nfapi_nr_vnf_param_req(config, p5_idx, &req);
return 0;
}
@@ -1800,21 +1809,10 @@ void vnf_nr_deallocate_p4_p5_vendor_ext(void *header) {
free(header);
}
nfapi_vnf_config_t *config = 0;
void vnf_nr_start_thread(void *ptr) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] VNF NFAPI thread - nfapi_vnf_start()%s\n", __FUNCTION__);
pthread_setname_np(pthread_self(), "VNF");
config = (nfapi_vnf_config_t *)ptr;
nfapi_nr_vnf_start(config);
}
void vnf_start_thread(void *ptr) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] VNF NFAPI thread - nfapi_vnf_start()%s\n", __FUNCTION__);
pthread_setname_np(pthread_self(), "VNF");
config = (nfapi_vnf_config_t *)ptr;
nfapi_vnf_start(config);
nfapi_vnf_start((nfapi_vnf_config_t *)ptr);
}
static vnf_info vnf;
@@ -1868,13 +1866,16 @@ void configure_nr_nfapi_vnf(char *vnf_addr, int vnf_p5_port, char *pnf_ip_addr,
config->codec_config.allocate = &vnf_nr_allocate;
config->codec_config.deallocate = &vnf_nr_deallocate;
memset(&UL_RCC_INFO, 0, sizeof(UL_RCC_IND_t));
#ifdef ENABLE_SOCKET
config->unpack_func = &nfapi_nr_p5_message_unpack;
config->hdr_unpack_func = &nfapi_nr_p5_message_header_unpack;
config->pack_func = &nfapi_nr_p5_message_pack;
config->send_p5_msg = &vnf_nr_send_p5_message;
config->send_p5_msg = &vnf_nr_send_p5_msg;
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Creating VNF NFAPI start thread %s\n", __FUNCTION__);
pthread_create(&vnf_start_pthread, NULL, (void *)&vnf_nr_start_thread, config);
pthread_create(&vnf_p5_init_and_receive_pthread, NULL, (void *)&vnf_start_p5_thread, config);
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Created VNF NFAPI start thread %s\n", __FUNCTION__);
#endif
}
@@ -1922,7 +1923,7 @@ void configure_nfapi_vnf(char *vnf_addr, int vnf_p5_port, char *pnf_ip_addr, int
config->codec_config.deallocate = &vnf_deallocate;
memset(&UL_RCC_INFO,0,sizeof(UL_RCC_IND_t));
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Creating VNF NFAPI start thread %s\n", __FUNCTION__);
pthread_create(&vnf_start_pthread, NULL, (void *)&vnf_start_thread, config);
pthread_create(&vnf_p5_init_and_receive_pthread, NULL, (void *)&vnf_start_thread, config);
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] Created VNF NFAPI start thread %s\n", __FUNCTION__);
}

1
nfapi/oai_integration/socket/CMakeLists.txt
View File

@@ -1,5 +1,6 @@
add_library(nfapi_socket_lib
socket_pnf.c
socket_vnf.c
socket_common.c
)

6
nfapi/oai_integration/socket/include/socket_common.h
View File

@@ -21,12 +21,18 @@
#ifndef SOCKET_COMMON_H
#define SOCKET_COMMON_H
#define _GNU_SOURCE /* required for pthread_getname_np */
#include <errno.h>
#include <stdbool.h>
#include <stdio.h>
#include <netinet/in.h>
#include <netinet/sctp.h>
#include <assertions.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/socket.h>
#include <pthread.h>
#include <netdb.h>
int socket_send_p5_msg(const int sctp,
const int socket,

34
nfapi/oai_integration/socket/include/socket_vnf.h Normal file
View File

@@ -0,0 +1,34 @@
/*
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1 (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.openairinterface.org/?page_id=698
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
* contact@openairinterface.org
*/
#ifndef SOCKET_VNF_H
#define SOCKET_VNF_H
#include "socket_common.h"
#include "vnf.h"
#include "vnf_p7.h"
#include "nr_fapi_p5.h"
#include "nr_nfapi_p7.h"
#include "nr_fapi_p7.h"
bool vnf_nr_send_p5_msg(vnf_t *vnf, uint16_t p5_idx, nfapi_nr_p4_p5_message_header_t* msg, uint32_t msg_len);
bool vnf_nr_send_p7_msg(vnf_p7_t *vnf_p7, nfapi_nr_p7_message_header_t* header);
void vnf_start_p5_thread(void* ptr);
void* vnf_nr_start_p7_thread(void* ptr);
#endif // SOCKET_VNF_H

861
nfapi/oai_integration/socket/socket_vnf.c Normal file
View File

@@ -0,0 +1,861 @@
/*
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1 (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.openairinterface.org/?page_id=698
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
* contact@openairinterface.org
*/
#include "socket_vnf.h"
#include "nfapi.h"
#include "nfapi/oai_integration/vendor_ext.h" //TODO: Remove this include when removing the Aerial transport stuff
static bool send_p5_msg(vnf_t *vnf, nfapi_vnf_pnf_info_t *pnf, const void *msg, int len, uint8_t stream)
{
int result = socket_send_p5_msg(vnf->sctp, pnf->p5_sock, &pnf->p5_pnf_sockaddr, msg, len, stream);
if (result != len) {
if (result < 0) {
// error
NFAPI_TRACE(NFAPI_TRACE_ERROR, "sctp sendto failed errno: %d\n", errno);
} else {
// did not send the entire message
}
}
return result == len;
}
bool vnf_nr_send_p5_msg(vnf_t *vnf, uint16_t p5_idx, nfapi_nr_p4_p5_message_header_t *msg, uint32_t msg_len)
{
nfapi_vnf_pnf_info_t *pnf = nfapi_vnf_pnf_list_find(&(vnf->_public), p5_idx);
if (pnf) {
// pack the message for transmission
int packedMessageLength = 0;
if (NFAPI_MODE == NFAPI_MODE_AERIAL) {
#ifdef ENABLE_AERIAL // TODO: Remove this into the aerial VNF transport module
// In case it is a FAPI message, create 2 messages, one with nFAPI header for OAI PNF and one with no nFAPI header for Aerial
// L1
// create FAPI tx_buffer
uint8_t tx_messagebufferFAPI[sizeof(vnf->tx_message_buffer)];
int packedMessageLengthFAPI = -1;
packedMessageLengthFAPI =
fapi_nr_p5_message_pack(msg, msg_len, tx_messagebufferFAPI, sizeof(tx_messagebufferFAPI), &vnf->_public.codec_config);
return aerial_send_P5_msg(tx_messagebufferFAPI, packedMessageLengthFAPI, msg);
#else
return 0;
#endif
} else {
packedMessageLength =
vnf->_public.pack_func(msg, msg_len, vnf->tx_message_buffer, sizeof(vnf->tx_message_buffer), &vnf->_public.codec_config);
if (packedMessageLength < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "nfapi_nr_p5_message_pack failed with return %d\n", packedMessageLength);
return false;
}
return send_p5_msg(vnf, pnf, vnf->tx_message_buffer, packedMessageLength, 0);
}
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() cannot find pnf info for p5_idx:%d\n", __FUNCTION__, p5_idx);
return false;
}
}
bool vnf_nr_send_p7_msg(vnf_p7_t *vnf_p7, nfapi_nr_p7_message_header_t *header)
{
nfapi_vnf_p7_connection_info_t *p7_connection = vnf_p7_connection_info_list_find(vnf_p7, header->phy_id);
if (p7_connection) {
int send_result = 0;
uint8_t buffer[1024 * 1024 * 3];
header->m_segment_sequence = NFAPI_NR_P7_SET_MSS(0, 0, p7_connection->sequence_number);
int len = vnf_p7->_public.pack_func(header, buffer, sizeof(buffer), &vnf_p7->_public.codec_config);
if (len < 0) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() failed to pack p7 message phy_id:%d\n", __FUNCTION__, header->phy_id);
return false;
}
if (len > vnf_p7->_public.segment_size) {
// todo : consider replacing with the sendmmsg call
// todo : worry about blocking writes?
// segmenting the transmit
int msg_body_len = len - NFAPI_NR_P7_HEADER_LENGTH;
int seg_body_len = vnf_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 tx_buffer[vnf_p7->_public.segment_size];
NFAPI_TRACE(NFAPI_TRACE_DEBUG,
"%s() MORE THAN ONE SEGMENT phy_id:%d nfapi_p7_message_pack()=len=%d vnf_p7->_public.segment_size:%u\n",
__FUNCTION__,
header->phy_id,
len,
vnf_p7->_public.segment_size);
for (segment = 0; segment < segment_count; ++segment) {
uint8_t last = 0;
uint16_t size = vnf_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(&tx_buffer[0], buffer, NFAPI_NR_P7_HEADER_LENGTH);
// set the segment length, update and push m_segment_sequence
uint8_t *buf_ptr = &tx_buffer[4];
uint8_t *buf_ptr_end = &tx_buffer[10];
push32(segment_size, (&buf_ptr), buf_ptr_end);
header->m_segment_sequence = NFAPI_NR_P7_SET_MSS((!last), segment, p7_connection->sequence_number);
push16(header->m_segment_sequence, (&buf_ptr), buf_ptr_end);
memcpy(&tx_buffer[NFAPI_NR_P7_HEADER_LENGTH], &buffer[0] + offset, size);
offset += size;
if (vnf_p7->_public.checksum_enabled) {
nfapi_nr_p7_update_checksum(tx_buffer, segment_size);
}
const uint32_t time =
calculate_transmit_timestamp(p7_connection->mu, p7_connection->sfn, p7_connection->slot, vnf_p7->slot_start_time_hr);
nfapi_nr_p7_update_transmit_timestamp(tx_buffer, time);
send_result = socket_send_p7_msg(vnf_p7->socket, &(p7_connection->remote_addr), &tx_buffer[0], segment_size);
}
} else {
if (vnf_p7->_public.checksum_enabled) {
nfapi_nr_p7_update_checksum(buffer, len);
}
const uint32_t time =
calculate_transmit_timestamp(p7_connection->mu, p7_connection->sfn, p7_connection->slot, vnf_p7->slot_start_time_hr);
nfapi_nr_p7_update_transmit_timestamp(buffer, time);
send_result = socket_send_p7_msg(vnf_p7->socket, &(p7_connection->remote_addr), &buffer[0], len);
}
p7_connection->sequence_number++;
return send_result == 0;
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() cannot find p7 connection info for phy_id:%d\n", __FUNCTION__, header->phy_id);
return false;
}
}
static int vnf_nr_read_dispatch_message(nfapi_vnf_config_t *config, nfapi_vnf_pnf_info_t *pnf)
{
if (1) {
int socket_connected = 1;
vnf_t *vnf = (vnf_t *)(config);
// 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 (vnf->sctp) {
message_size =
sctp_recvmsg(pnf->p5_sock, header_buffer, header_buffer_size, (struct sockaddr *)&addr, &addr_len, &sndrcvinfo, &flags);
} else {
message_size = recv(pnf->p5_sock, header_buffer, header_buffer_size, flags);
}
if (message_size == -1) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Failed to peek sctp message size errno:%d\n", errno);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Peeked message with length :0x%02x\n", message_size);
for (int i = 0; i < message_size; i++) {
printf("%02x ", header_buffer[i]);
}
printf("\n");
nfapi_nr_p4_p5_message_header_t header;
const bool result = config->hdr_unpack_func(header_buffer, header_buffer_size, &header, sizeof(header), 0);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Failed to decode message header\n");
return 0;
}
message_size = header.message_length + header_buffer_size;
// now have the size of the mesage
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF After header unpacking msg size is :0x%02x\n", message_size);
}
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, "VNF 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 (vnf->sctp) {
recvmsg_result =
sctp_recvmsg(pnf->p5_sock, read_buffer, message_size, (struct sockaddr *)&addr, &addr_len, &sndrcvinfo, &flags);
} else {
recvmsg_result = recv(pnf->p5_sock, read_buffer, message_size, 0);
}
if (recvmsg_result == -1) {
int tmp = errno;
NFAPI_TRACE(NFAPI_TRACE_INFO, "Failed to read sctp message size error %s errno:%d\n", strerror(tmp), tmp);
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF after recv flags are :0x%02x\n", flags);
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 || !vnf->sctp) {
vnf_nr_handle_p4_p5_message(read_buffer, message_size, pnf->p5_idx, config);
} else {
int tmp = errno;
NFAPI_TRACE(NFAPI_TRACE_WARN,
"sctp_recvmsg: unhandled mode with flags 0x%x and error %s errno:%d\n",
flags,
strerror(tmp),
tmp);
// assume socket disconnected
NFAPI_TRACE(NFAPI_TRACE_WARN, "Disconnected socket\n");
socket_connected = 0;
}
}
}
}
if (dynamic_buffer) {
free(dynamic_buffer);
}
return socket_connected;
}
}
static int nfapi_nr_vnf_p5_start(nfapi_vnf_config_t *config)
{
// Verify that config is not null
if (config == 0)
return -1;
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
int p5ListenSock, p5Sock;
struct sockaddr_in addr = {0};
socklen_t addrSize;
struct sockaddr_in6 addr6 = {0};
struct sctp_event_subscribe events = {0};
struct sctp_initmsg initMsg = {0};
int noDelay;
vnf_t *vnf = (vnf_t *)(config);
NFAPI_TRACE(NFAPI_TRACE_INFO, "Starting P5 VNF connection on port %u\n", config->vnf_p5_port);
{
int protocol;
int domain;
if (vnf->sctp)
protocol = IPPROTO_SCTP;
else
protocol = IPPROTO_IP;
if (config->vnf_ipv6) {
domain = PF_INET6;
} else {
domain = AF_INET;
}
// open the SCTP socket
if ((p5ListenSock = socket(domain, SOCK_STREAM, protocol)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P5 socket errno: %d\n", errno);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "P5 socket created... %d\n", p5ListenSock);
}
if (vnf->sctp) {
// configure for MSG_NOTIFICATION
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_EVENTS, &events, sizeof(struct sctp_event_subscribe)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (SCTP_EVENTS) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_NOTE, "VNF Setting the SCTP_INITMSG\n");
// configure the SCTP socket options
initMsg.sinit_num_ostreams = 5; // MAX_SCTP_STREAMS; // number of output streams can be greater
initMsg.sinit_max_instreams = 5; // MAX_SCTP_STREAMS; // number of output streams can be greater
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_INITMSG, &initMsg, sizeof(initMsg)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (SCTP_INITMSG) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
noDelay = 1;
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_NODELAY, &noDelay, sizeof(noDelay)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (STCP_NODELAY) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
struct sctp_event_subscribe events;
memset((void *)&events, 0, sizeof(events));
events.sctp_data_io_event = 1;
if (setsockopt(p5ListenSock, SOL_SCTP, SCTP_EVENTS, (const void *)&events, sizeof(events)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt errno: %d\n", errno);
close(p5ListenSock);
return -1;
}
} else {
int error;
socklen_t len = sizeof(error);
if (getsockopt(p5ListenSock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After getsockopt errno: %d\n", errno);
close(p5ListenSock);
return -1;
}
}
if (config->vnf_ipv6) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "IPV6 binding to port %d %d\n", config->vnf_p5_port, p5ListenSock);
addr6.sin6_family = AF_INET6;
addr6.sin6_port = htons(config->vnf_p5_port);
addr6.sin6_addr = in6addr_any;
// bind to the configured address and port
if (bind(p5ListenSock, (struct sockaddr *)&addr6, sizeof(struct sockaddr_in6)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
} else if (config->vnf_ipv4) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "IPV4 binding to port %d\n", config->vnf_p5_port);
addr.sin_family = AF_INET;
addr.sin_port = htons(config->vnf_p5_port);
addr.sin_addr.s_addr = INADDR_ANY;
// bind to the configured address and port
if (bind(p5ListenSock, (struct sockaddr *)&addr, sizeof(struct sockaddr_in)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
close(p5ListenSock);
AssertFatal(1 == 0, "Failed to bind socket with errno: %d\n", errno);
return 0;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "bind succeeded..%d.\n", p5ListenSock);
// put the socket into listen mode
if (listen(p5ListenSock, 2) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After listen errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "listen succeeded...\n");
struct timeval tv;
fd_set read_fd_set;
int p5_idx = 0;
while (vnf->terminate == 0) {
FD_ZERO(&read_fd_set);
FD_SET(p5ListenSock, &read_fd_set);
int max_fd = p5ListenSock;
tv.tv_sec = 5;
tv.tv_usec = 0;
nfapi_vnf_pnf_info_t *pnf = config->pnf_list;
while (pnf != 0) {
if (pnf->connected) {
FD_SET(pnf->p5_sock, &read_fd_set);
if (pnf->p5_sock > max_fd) {
max_fd = pnf->p5_sock;
}
}
pnf = pnf->next;
}
int select_result = select(max_fd + 1, &read_fd_set, 0, 0, &tv);
if (select_result == -1) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "select result %d errno %d\n", select_result, errno);
close(p5ListenSock);
return 0;
} else if (select_result) {
if (FD_ISSET(p5ListenSock, &read_fd_set)) {
addrSize = sizeof(struct sockaddr_in);
NFAPI_TRACE(NFAPI_TRACE_INFO, "Accepting connection from PNF...\n");
p5Sock = accept(p5ListenSock, (struct sockaddr *)&addr, &addrSize);
if (p5Sock < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Failed to accept PNF connection reason:%d\n", errno);
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO,
"PNF connection (fd:%d) accepted from %s:%d \n",
p5Sock,
inet_ntoa(addr.sin_addr),
ntohs(addr.sin_port));
nfapi_vnf_pnf_info_t *pnf = (nfapi_vnf_pnf_info_t *)malloc(sizeof(nfapi_vnf_pnf_info_t));
NFAPI_TRACE(NFAPI_TRACE_INFO, "MALLOC nfapi_vnf_pnf_info_t for pnf_list pnf:%p\n", pnf);
memset(pnf, 0, sizeof(nfapi_vnf_pnf_info_t));
pnf->p5_sock = p5Sock;
pnf->p5_idx = p5_idx++;
pnf->p5_pnf_sockaddr = addr;
pnf->connected = 1;
nfapi_vnf_pnf_list_add(config, pnf);
// Inform mac that a pnf connection has been established
// todo : allow mac to 'accept' the connection. i.e. to
// reject it.
if (config->pnf_nr_connection_indication != 0) {
(config->pnf_nr_connection_indication)(config, pnf->p5_idx);
}
// check the connection status
if (vnf->sctp) {
struct sctp_status status;
(void)memset(&status, 0, sizeof(struct sctp_status));
socklen_t optLen = (socklen_t)sizeof(struct sctp_status);
if (getsockopt(p5Sock, IPPROTO_SCTP, SCTP_STATUS, &status, &optLen) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After getsockopt errno: %d\n", errno);
return -1;
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Association ID = %d\n", status.sstat_assoc_id);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Receiver window size = %d\n", status.sstat_rwnd);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF In Streams = %d\n", status.sstat_instrms);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Out Streams = %d\n", status.sstat_outstrms);
}
}
}
} else {
uint8_t delete_pnfs = 0;
nfapi_vnf_pnf_info_t *pnf = config->pnf_list;
while (pnf != 0) {
if (FD_ISSET(pnf->p5_sock, &read_fd_set)) {
if (vnf_nr_read_dispatch_message(config, pnf) == 0) {
if (config->pnf_disconnect_indication != 0) {
(config->pnf_disconnect_indication)(config, pnf->p5_idx);
}
close(pnf->p5_sock);
pnf->to_delete = 1;
delete_pnfs = 1;
}
}
pnf = pnf->next;
}
if (delete_pnfs) {
nfapi_vnf_pnf_info_t *pnf = config->pnf_list;
nfapi_vnf_pnf_info_t *prev = 0;
while (pnf != 0) {
nfapi_vnf_pnf_info_t *curr = pnf;
if (pnf->to_delete == 1) {
if (prev == 0) {
config->pnf_list = pnf->next;
} else {
prev->next = pnf->next;
}
pnf = pnf->next;
free(curr);
} else {
prev = pnf;
pnf = pnf->next;
}
}
}
}
continue;
} else {
// timeout
// Should we test for socket closure here every second?
continue;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p5Sock socket's\n");
{
nfapi_vnf_pnf_info_t *curr = config->pnf_list;
while (curr != NULL) {
if (config->pnf_disconnect_indication) {
(config->pnf_disconnect_indication)(config, curr->p5_idx);
}
close(curr->p5_sock);
curr = curr->next;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p5Listen socket\n");
close(p5ListenSock);
return 0;
}
void vnf_start_p5_thread(void *ptr)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "[VNF] VNF NFAPI thread - nfapi_vnf_start()%s\n", __FUNCTION__);
pthread_setname_np(pthread_self(), "VNF");
nfapi_nr_vnf_p5_start((nfapi_vnf_config_t *)ptr);
}
void vnf_nr_reassemble_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t *vnf_p7)
{
nfapi_nr_p7_message_header_t messageHeader;
// validate the input params
if (pRecvMsg == NULL || recvMsgLen < 4 || vnf_p7 == NULL) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "vnf_handle_p7_message: invalid input params (%p %d %p)\n", pRecvMsg, recvMsgLen, vnf_p7);
return;
}
// unpack the message header
const bool result =
vnf_p7->_public.hdr_unpack_func(pRecvMsg, recvMsgLen, &messageHeader, sizeof(messageHeader), &vnf_p7->_public.codec_config);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Unpack message header failed, ignoring\n");
return;
}
if (vnf_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 msg:%d len:%d\n",
checksum,
messageHeader.checksum,
messageHeader.message_id,
recvMsgLen);
return;
}
}
uint8_t m = NFAPI_P7_GET_MORE(messageHeader.m_segment_sequence);
uint8_t segment_num = NFAPI_P7_GET_SEGMENT(messageHeader.m_segment_sequence);
uint8_t sequence_num = NFAPI_P7_GET_SEQUENCE(messageHeader.m_segment_sequence);
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;
}
// vnf_dispatch_p7_message(&messageHeader, pRecvMsg, recvMsgLen, vnf_p7);
vnf_nr_handle_p7_message(pRecvMsg, recvMsgLen, vnf_p7);
} else {
nfapi_vnf_p7_connection_info_t *phy = vnf_p7_connection_info_list_find(vnf_p7, messageHeader.phy_id);
if (phy) {
vnf_p7_rx_message_t *rx_msg = vnf_p7_rx_reassembly_queue_add_segment(vnf_p7,
&(phy->reassembly_queue),
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;
uint16_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 (phy->reassembly_buffer_size < length) {
vnf_p7_free(vnf_p7, phy->reassembly_buffer);
phy->reassembly_buffer = 0;
}
if (phy->reassembly_buffer == 0) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Resizing VNF_P7 Reassembly buffer %d->%d\n", phy->reassembly_buffer_size, length);
phy->reassembly_buffer = (uint8_t *)vnf_p7_malloc(vnf_p7, length);
if (phy->reassembly_buffer == 0) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Failed to allocate VNF_P7 reassemby buffer len:%d\n", length);
return;
}
memset(phy->reassembly_buffer, 0, length);
phy->reassembly_buffer_size = length;
}
uint16_t offset = 0;
for (i = 0; i < rx_msg->num_segments_expected; ++i) {
if (i == 0) {
memcpy(phy->reassembly_buffer, rx_msg->segments[i].buffer, rx_msg->segments[i].length);
offset += rx_msg->segments[i].length;
} else {
memcpy(phy->reassembly_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;
}
}
vnf_nr_handle_p7_message(phy->reassembly_buffer, length, vnf_p7);
// delete the structure
vnf_p7_rx_reassembly_queue_remove_msg(vnf_p7, &(phy->reassembly_queue), rx_msg);
}
// see corresponding comment in pnf_nr_handle_p7_message() [same commit]
vnf_p7_rx_reassembly_queue_remove_old_msgs(vnf_p7, &(phy->reassembly_queue), 10000);
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "Unknown phy id %d\n", messageHeader.phy_id);
}
}
}
int vnf_nr_p7_read_dispatch_message(vnf_p7_t *vnf_p7)
{
int recvfrom_result = 0;
struct sockaddr_in remote_addr;
socklen_t remote_addr_size = sizeof(remote_addr);
do {
// peek the header
uint8_t header_buffer[NFAPI_NR_P7_HEADER_LENGTH];
recvfrom_result = recvfrom(vnf_p7->socket,
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;
const bool result = vnf_p7->_public.hdr_unpack_func(header_buffer, NFAPI_NR_P7_HEADER_LENGTH, &header, sizeof(header), 0);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Unpack message header failed, ignoring\n");
return -1;
}
// resize the buffer if we have a large segment
if (header.message_length > vnf_p7->rx_message_buffer_size) {
NFAPI_TRACE(NFAPI_TRACE_NOTE, "reallocing rx buffer %d\n", header.message_length);
vnf_p7->rx_message_buffer = realloc(vnf_p7->rx_message_buffer, header.message_length);
vnf_p7->rx_message_buffer_size = header.message_length;
}
// read the segment
recvfrom_result = recvfrom(vnf_p7->socket,
vnf_p7->rx_message_buffer,
header.message_length,
MSG_WAITALL | MSG_TRUNC,
(struct sockaddr *)&remote_addr,
&remote_addr_size);
NFAPI_TRACE(NFAPI_TRACE_DEBUG, "recvfrom_result = %d from %s():%d\n", recvfrom_result, __FUNCTION__, __LINE__);
// todo : how to handle incomplete readfroms, need some sort of buffer/select
if (recvfrom_result > 0) {
if (recvfrom_result != header.message_length) {
NFAPI_TRACE(NFAPI_TRACE_ERROR,
"(%d) Received unexpected number of bytes. %d != %d",
__LINE__,
recvfrom_result,
header.message_length);
break;
}
NFAPI_TRACE(NFAPI_TRACE_DEBUG, "Calling vnf_nr_reassemble_p7_message from %d\n", __LINE__);
vnf_nr_reassemble_p7_message(vnf_p7->rx_message_buffer, recvfrom_result, vnf_p7);
return 0;
} else {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "recvfrom failed %d %d\n", recvfrom_result, errno);
}
}
if (recvfrom_result == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// return to the select
} else {
NFAPI_TRACE(NFAPI_TRACE_WARN, "%s recvfrom failed errno:%d\n", __FUNCTION__, errno);
}
}
} while (recvfrom_result > 0);
return 0;
}
static int nfapi_nr_vnf_p7_start(nfapi_vnf_p7_config_t *config)
{
if (config == 0)
return -1;
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
vnf_p7_t *vnf_p7 = (vnf_p7_t *)config;
// Create p7 receive udp port
// todo : this needs updating for Ipv6
NFAPI_TRACE(NFAPI_TRACE_INFO, "Initialising VNF P7 port:%u\n", config->port);
// open the UDP socket
if ((vnf_p7->socket = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P7 socket errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 socket created...\n");
// configure the UDP socket options
int iptos_value = 0;
if (setsockopt(vnf_p7->socket, IPPROTO_IP, IP_TOS, &iptos_value, sizeof(iptos_value)) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (IP_TOS) errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 setsockopt succeeded...\n");
// Create the address structure
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(config->port);
addr.sin_addr.s_addr = INADDR_ANY;
// bind to the configured port
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 binding too %s:%d\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
if (bind(vnf_p7->socket, (struct sockaddr *)&addr, sizeof(struct sockaddr_in)) < 0)
// if (sctp_bindx(config->socket, (struct sockaddr *)&addr, sizeof(struct sockaddr_in), 0) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 bind succeeded...\n");
// struct timespec original_pselect_timeout;
struct timespec pselect_timeout;
pselect_timeout.tv_sec = 100;
pselect_timeout.tv_nsec = 0;
struct timespec ref_time;
clock_gettime(CLOCK_MONOTONIC, &ref_time);
while (vnf_p7->terminate == 0) {
fd_set rfds;
int maxSock = 0;
FD_ZERO(&rfds);
int selectRetval = 0;
// Add the p7 socket
FD_SET(vnf_p7->socket, &rfds);
maxSock = vnf_p7->socket;
selectRetval = pselect(maxSock + 1, &rfds, NULL, NULL, &pselect_timeout, NULL);
if (selectRetval == 0) {
// pselect timed out, continue
} else if (selectRetval > 0) {
// have a p7 message
if (FD_ISSET(vnf_p7->socket, &rfds)) {
vnf_nr_p7_read_dispatch_message(vnf_p7);
}
} else {
// pselect error
if (selectRetval == -1 && errno == EINTR) {
// a sigal was received.
} else {
// should we exit now?
if (selectRetval == -1 && errno == 22) // invalid argument??? not sure about timeout duration
{
usleep(100000);
}
}
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p7 socket\n");
close(vnf_p7->socket);
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() returning\n", __FUNCTION__);
return 0;
}
void *vnf_nr_start_p7_thread(void *ptr)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
pthread_setname_np(pthread_self(), "VNF_P7");
nfapi_vnf_p7_config_t *config = (nfapi_vnf_p7_config_t *)ptr;
nfapi_nr_vnf_p7_start(config);
return config;
}

6
nfapi/open-nFAPI/vnf/inc/vnf.h
View File

@@ -35,16 +35,18 @@ struct vnf_t
int vnf_pack_and_send_p5_message(vnf_t* vnf, uint16_t p5_idx, nfapi_p4_p5_message_header_t* msg, uint16_t msg_len);
bool vnf_nr_send_p5_message(vnf_t* vnf, uint16_t p5_idx, nfapi_nr_p4_p5_message_header_t* msg, uint32_t msg_len);
int vnf_pack_and_send_p4_message(vnf_t* vnf, uint16_t p5_idx, nfapi_p4_p5_message_header_t* msg, uint16_t msg_len);
int vnf_read_dispatch_message(nfapi_vnf_config_t* config, nfapi_vnf_pnf_info_t* pnf);
int vnf_nr_read_dispatch_message(nfapi_vnf_config_t* config, nfapi_vnf_pnf_info_t* pnf);
void nfapi_vnf_phy_info_list_add(nfapi_vnf_config_t* config, nfapi_vnf_phy_info_t* info);
nfapi_vnf_phy_info_t* nfapi_vnf_phy_info_list_find(nfapi_vnf_config_t* config, uint16_t phy_id);
void nfapi_vnf_pnf_list_add(nfapi_vnf_config_t* config, nfapi_vnf_pnf_info_t* node);
nfapi_vnf_pnf_info_t* nfapi_vnf_pnf_list_find(nfapi_vnf_config_t* config, int p5_idx);
void vnf_nr_handle_p4_p5_message(void *pRecvMsg, int recvMsgLen, int p5_idx, nfapi_vnf_config_t* config);
#endif // _VNF_H_

12
nfapi/open-nFAPI/vnf/inc/vnf_p7.h
View File

@@ -19,7 +19,6 @@
#define _VNF_P7_H_
#include "nfapi_vnf_interface.h"
#define TIMEHR_SEC(_time_hr) ((uint32_t)(_time_hr) >> 20)
#define TIMEHR_USEC(_time_hr) ((uint32_t)(_time_hr) & 0xFFFFF)
#define TIME2TIMEHR(_time) (((uint32_t)(_time.tv_sec) & 0xFFF) << 20 | ((uint32_t)(_time.tv_usec) & 0xFFFFF))
@@ -132,15 +131,20 @@ int vnf_nr_sync(vnf_p7_t* vnf_p7, nfapi_vnf_p7_connection_info_t* p7_info);
int send_mac_subframe_indications(vnf_p7_t* config);
int send_mac_slot_indications(vnf_p7_t* config);
int vnf_p7_read_dispatch_message(vnf_p7_t* vnf_p7 );
int vnf_nr_p7_read_dispatch_message(vnf_p7_t* vnf_p7 );
void vnf_nr_handle_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7);
vnf_p7_rx_message_t* vnf_p7_rx_reassembly_queue_add_segment(vnf_p7_t* vnf_p7, vnf_p7_rx_reassembly_queue_t* queue, uint16_t sequence_number, uint16_t segment_number, uint8_t m, uint8_t* data, uint16_t data_len);
void* vnf_p7_malloc(vnf_p7_t* vnf_p7, size_t size);
void vnf_p7_free(vnf_p7_t* vnf_p7, void* ptr);
void vnf_p7_rx_reassembly_queue_remove_msg(vnf_p7_t* vnf_p7, vnf_p7_rx_reassembly_queue_t* queue, vnf_p7_rx_message_t* msg);
void vnf_p7_rx_reassembly_queue_remove_old_msgs(vnf_p7_t* vnf_p7, vnf_p7_rx_reassembly_queue_t* queue, uint32_t delta);
uint32_t calculate_transmit_timestamp(int mu, uint16_t sfn, uint16_t slot, uint32_t slot_start_time_hr);
void vnf_p7_connection_info_list_add(vnf_p7_t* vnf_p7, nfapi_vnf_p7_connection_info_t* node);
nfapi_vnf_p7_connection_info_t* vnf_p7_connection_info_list_find(vnf_p7_t* vnf_p7, uint16_t phy_id);
nfapi_vnf_p7_connection_info_t* vnf_p7_connection_info_list_delete(vnf_p7_t* vnf_p7, uint16_t phy_id);
int vnf_p7_pack_and_send_p7_msg(vnf_p7_t* vnf_p7, nfapi_p7_message_header_t* header);
bool vnf_nr_send_p7_msg(vnf_p7_t* vnf_p7, nfapi_nr_p7_message_header_t* header);
void vnf_p7_release_msg(vnf_p7_t* vnf_p7, nfapi_p7_message_header_t* header);
void vnf_p7_release_pdu(vnf_p7_t* vnf_p7, void* pdu);

3
nfapi/open-nFAPI/vnf/public_inc/nfapi_vnf_interface.h
View File

@@ -529,7 +529,6 @@ void nfapi_vnf_config_destory(nfapi_vnf_config_t* config);
*
* This function will not return untill nfapi_vnf_stop is called
*/
int nfapi_nr_vnf_start(nfapi_vnf_config_t* config);
int nfapi_vnf_start(nfapi_vnf_config_t* config);
@@ -1000,7 +999,7 @@ void nfapi_vnf_p7_config_destory(nfapi_vnf_p7_config_t* config);
* function is called.
*/
int nfapi_vnf_p7_start(nfapi_vnf_p7_config_t* config);
int nfapi_nr_vnf_p7_start(nfapi_vnf_p7_config_t* config);
/*! Stop the VNF P7 library.

187
nfapi/open-nFAPI/vnf/src/vnf.c
View File

@@ -1267,146 +1267,6 @@ void vnf_handle_p4_p5_message(void *pRecvMsg, int recvMsgLen, int p5_idx, nfapi_
}
}
int vnf_nr_read_dispatch_message(nfapi_vnf_config_t* config, nfapi_vnf_pnf_info_t* pnf)
{
if (1) {
int socket_connected = 1;
vnf_t *vnf = (vnf_t *)(config);
// 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 (vnf->sctp) {
message_size =
sctp_recvmsg(pnf->p5_sock,
header_buffer,
header_buffer_size,
(struct sockaddr *)&addr,
&addr_len,
&sndrcvinfo,
&flags);
} else {
message_size = recv(pnf->p5_sock,
header_buffer,
header_buffer_size,
flags);
}
if (message_size == -1) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Failed to peek sctp message size errno:%d\n", errno);
return 0;
}
nfapi_nr_p4_p5_message_header_t header;
const bool result = config->hdr_unpack_func(header_buffer, header_buffer_size, &header, sizeof(header), 0);
if (!result) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Failed to decode 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, "VNF 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 (vnf->sctp) {
recvmsg_result = sctp_recvmsg(pnf->p5_sock,
read_buffer,
message_size,
(struct sockaddr *)&addr,
&addr_len,
&sndrcvinfo,
&flags);
} else {
recvmsg_result = recv(pnf->p5_sock,
read_buffer,
message_size,
0);
}
if (recvmsg_result == -1) {
NFAPI_TRACE(NFAPI_TRACE_INFO, "Failed to read sctp message size errno:%d\n", errno);
} else {
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 || !vnf->sctp) {
// printf("\nVNF RECEIVES:\n");
// for(int i=0; i<message_size; i++){
// printf("%d", read_buffer[i]);
// }
// printf("\n");
vnf_nr_handle_p4_p5_message(read_buffer, message_size, pnf->p5_idx, config);
} 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 vnf_read_dispatch_message(nfapi_vnf_config_t* config, nfapi_vnf_pnf_info_t* pnf)
{
@@ -1581,53 +1441,6 @@ static int vnf_send_p5_msg(vnf_t *vnf, nfapi_vnf_pnf_info_t *pnf, const void *ms
return 0;
}
bool vnf_nr_send_p5_message(vnf_t* vnf, uint16_t p5_idx, nfapi_nr_p4_p5_message_header_t* msg, uint32_t msg_len)
{
nfapi_vnf_pnf_info_t* pnf = nfapi_vnf_pnf_list_find(&(vnf->_public), p5_idx);
if(pnf)
{
// pack the message for transmission
int packedMessageLength = 0;
if (NFAPI_MODE == NFAPI_MODE_AERIAL) {
#ifdef ENABLE_AERIAL
// In case it is a FAPI message, create 2 messages, one with nFAPI header for OAI PNF and one with no nFAPI header for Aerial
// L1
// create FAPI tx_buffer
uint8_t tx_messagebufferFAPI[sizeof(vnf->tx_message_buffer)];
int packedMessageLengthFAPI = -1;
packedMessageLengthFAPI =
fapi_nr_p5_message_pack(msg, msg_len, tx_messagebufferFAPI, sizeof(tx_messagebufferFAPI), &vnf->_public.codec_config);
return aerial_send_P5_msg(tx_messagebufferFAPI, packedMessageLengthFAPI, msg) == 0;
#else
return true;
#endif
} else {
packedMessageLength = nfapi_nr_p5_message_pack(msg,
msg_len,
vnf->tx_message_buffer,
sizeof(vnf->tx_message_buffer),
&vnf->_public.codec_config);
if (packedMessageLength < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "nfapi_nr_p5_message_pack failed with return %d\n", packedMessageLength);
return false;
}
// printf("msg id = 0x%02x, entire message length: %d\n", msg->message_id, packedMessageLength);
// for (int i = 0; i < packedMessageLength; i++) {
// printf(" msg->msg_buf[%d] = 0x%02x\n", i, ((uint8_t *) vnf->tx_message_buffer)[i]);
// }
return vnf_send_p5_msg(vnf, pnf, vnf->tx_message_buffer, packedMessageLength, 0);
}
} else {
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() cannot find pnf info for p5_idx:%d\n", __FUNCTION__, p5_idx);
return false;
}
}
int vnf_pack_and_send_p5_message(vnf_t* vnf, uint16_t p5_idx, nfapi_p4_p5_message_header_t* msg, uint16_t msg_len)
{
nfapi_vnf_pnf_info_t* pnf = nfapi_vnf_pnf_list_find(&(vnf->_public), p5_idx);

416
nfapi/open-nFAPI/vnf/src/vnf_interface.c
View File

@@ -65,422 +65,6 @@ void nfapi_vnf_config_destory(nfapi_vnf_config_t* config)
free(config);
}
int nfapi_nr_vnf_start(nfapi_vnf_config_t* config)
{
// Verify that config is not null
if(config == 0)
return -1;
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
int p5ListenSock, p5Sock;
struct sockaddr_in addr;
socklen_t addrSize;
struct sockaddr_in6 addr6;
struct sctp_event_subscribe events;
struct sctp_initmsg initMsg;
int noDelay;
(void)memset(&addr, 0, sizeof(struct sockaddr_in));
(void)memset(&addr6, 0, sizeof(struct sockaddr_in6));
(void)memset(&events, 0, sizeof(struct sctp_event_subscribe));
(void)memset(&initMsg, 0, sizeof(struct sctp_initmsg));
vnf_t* vnf = (vnf_t*)(config);
NFAPI_TRACE(NFAPI_TRACE_INFO, "Starting P5 VNF connection on port %u\n", config->vnf_p5_port);
/*
char * host = 0;
char * port = "4242";
struct addrinfo hints;
bzero(&hints, sizeof(struct addrinfo));
//hints.ai_flags=AI_PASSIVE;
//hints.ai_flags=AI_DEFAULT;
hints.ai_family=AF_UNSPEC;
//hints.ai_family=AF_INET6;
hints.ai_socktype=SOCK_STREAM;
//hints.ai_protocol=IPPROTO_SCTP
struct addrinfo *aiHead = 0;
int result = getaddrinfo(host, port, &hints, &aiHead);
NFAPI_TRACE(NFAPI_TRACE_INFO, "getaddrinfo return %d %d\n", result, errno);
while(aiHead->ai_next != NULL)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "addr info %d (IP %d UDP %d SCTP %d)\n %d (%d)\n",
aiHead->ai_protocol, IPPROTO_IP, IPPROTO_UDP, IPPROTO_SCTP,
aiHead->ai_flags, AI_PASSIVE);
char hostBfr[ NI_MAXHOST ];
char servBfr[ NI_MAXSERV ];
getnameinfo(aiHead->ai_addr,
aiHead->ai_addrlen,
hostBfr,
sizeof( hostBfr ),
servBfr,
sizeof( servBfr ),
NI_NUMERICHOST | NI_NUMERICSERV );
switch(aiHead->ai_family)
{
case PF_INET:
{
struct sockaddr_in *pSadrIn = (struct sockaddr_in*) aiHead->ai_addr;
printf(
" ai_addr = sin_family: %d (AF_INET = %d, "
"AF_INET6 = %d)\n"
" sin_addr: %s\n"
" sin_port: %s\n",
pSadrIn->sin_family,
AF_INET,
AF_INET6,
hostBfr,
servBfr );
}
break;
case PF_INET6:
{
struct sockaddr_in6 *pSadrIn6 = (struct sockaddr_in6*) aiHead->ai_addr;
fprintf( stderr,
" ai_addr = sin6_family: %d (AF_INET = %d, "
"AF_INET6 = %d) \n"
" sin6_addr: %s\n"
" sin6_port: %s\n"
" sin6_flowinfo: %d\n"
" sin6_scope_id: %d\n",
pSadrIn6->sin6_family,
AF_INET,
AF_INET6,
hostBfr,
servBfr,
pSadrIn6->sin6_flowinfo,
pSadrIn6->sin6_scope_id);
}
break;
default:
NFAPI_TRACE(NFAPI_TRACE_INFO, "Not ment to be here\n");
break;
}
aiHead = aiHead->ai_next;
}
*/
{
int protocol;
int domain;
if (vnf->sctp)
protocol = IPPROTO_SCTP;
else
protocol = IPPROTO_IP;
if(config->vnf_ipv6)
{
domain = PF_INET6;
}
else
{
domain = AF_INET;
}
// open the SCTP socket
if ((p5ListenSock = socket(domain, SOCK_STREAM, protocol)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P5 socket errno: %d\n", errno);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "P5 socket created... %d\n", p5ListenSock);
}
if (vnf->sctp)
{
// configure for MSG_NOTIFICATION
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_EVENTS, &events, sizeof(struct sctp_event_subscribe)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (SCTP_EVENTS) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_NOTE, "VNF Setting the SCTP_INITMSG\n");
// configure the SCTP socket options
initMsg.sinit_num_ostreams = 5; //MAX_SCTP_STREAMS; // number of output streams can be greater
initMsg.sinit_max_instreams = 5; //MAX_SCTP_STREAMS; // number of output streams can be greater
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_INITMSG, &initMsg, sizeof(initMsg)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (SCTP_INITMSG) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
noDelay = 1;
if (setsockopt(p5ListenSock, IPPROTO_SCTP, SCTP_NODELAY, &noDelay, sizeof(noDelay)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (STCP_NODELAY) errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
struct sctp_event_subscribe events;
memset( (void *)&events, 0, sizeof(events) );
events.sctp_data_io_event = 1;
if(setsockopt(p5ListenSock, SOL_SCTP, SCTP_EVENTS, (const void *)&events, sizeof(events)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt errno: %d\n", errno);
close(p5ListenSock);
return -1;
}
} else {
int error;
socklen_t len = sizeof(error);
if (getsockopt(p5ListenSock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After getsockopt errno: %d\n", errno);
close(p5ListenSock);
return -1;
}
}
if(config->vnf_ipv6)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "IPV6 binding to port %d %d\n", config->vnf_p5_port, p5ListenSock);
addr6.sin6_family = AF_INET6;
addr6.sin6_port = htons(config->vnf_p5_port);
addr6.sin6_addr = in6addr_any;
// bind to the configured address and port
if (bind(p5ListenSock, (struct sockaddr *)&addr6, sizeof(struct sockaddr_in6)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
}
else if(config->vnf_ipv4)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "IPV4 binding to port %d\n", config->vnf_p5_port);
addr.sin_family = AF_INET;
addr.sin_port = htons(config->vnf_p5_port);
addr.sin_addr.s_addr = INADDR_ANY;
// bind to the configured address and port
if (bind(p5ListenSock, (struct sockaddr *)&addr, sizeof(struct sockaddr_in)) < 0)
//if (sctp_bindx(p5ListenSock, (struct sockaddr *)&addr, sizeof(struct sockaddr_in), SCTP_BINDX_ADD_ADDR) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "bind succeeded..%d.\n", p5ListenSock);
// put the socket into listen mode
if (listen(p5ListenSock, 2) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After listen errno: %d\n", errno);
close(p5ListenSock);
return 0;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "listen succeeded...\n");
struct timeval tv;
fd_set read_fd_set;
int p5_idx = 0;
while(vnf->terminate == 0)
{
FD_ZERO(&read_fd_set);
FD_SET(p5ListenSock, &read_fd_set);
int max_fd = p5ListenSock;
tv.tv_sec = 5;
tv.tv_usec = 0;
nfapi_vnf_pnf_info_t* pnf = config->pnf_list;
while(pnf != 0)
{
if(pnf->connected)
{
FD_SET(pnf->p5_sock, &read_fd_set);
if (pnf->p5_sock > max_fd)
{
max_fd = pnf->p5_sock;
}
}
pnf = pnf->next;
}
int select_result = select(max_fd + 1, &read_fd_set, 0, 0, &tv);
if(select_result == -1)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "select result %d errno %d\n", select_result, errno);
close(p5ListenSock);
return 0;
}
else if(select_result)
{
if(FD_ISSET(p5ListenSock, &read_fd_set))
{
addrSize = sizeof(struct sockaddr_in);
NFAPI_TRACE(NFAPI_TRACE_INFO, "Accepting connection from PNF...\n");
p5Sock = accept(p5ListenSock, (struct sockaddr *)&addr, &addrSize);
if (p5Sock < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Failed to accept PNF connection reason:%d\n", errno);
}
else
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "PNF connection (fd:%d) accepted from %s:%d \n", p5Sock, inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
nfapi_vnf_pnf_info_t* pnf = (nfapi_vnf_pnf_info_t*)malloc(sizeof(nfapi_vnf_pnf_info_t));
NFAPI_TRACE(NFAPI_TRACE_INFO, "MALLOC nfapi_vnf_pnf_info_t for pnf_list pnf:%p\n", pnf);
memset(pnf, 0, sizeof(nfapi_vnf_pnf_info_t));
pnf->p5_sock = p5Sock;
pnf->p5_idx = p5_idx++;
pnf->p5_pnf_sockaddr = addr;
pnf->connected = 1;
nfapi_vnf_pnf_list_add(config, pnf);
// Inform mac that a pnf connection has been established
// todo : allow mac to 'accept' the connection. i.e. to
// reject it.
if(config->pnf_nr_connection_indication != 0)
{
(config->pnf_nr_connection_indication)(config, pnf->p5_idx);
}
// check the connection status
if (vnf->sctp)
{
struct sctp_status status;
(void)memset(&status, 0, sizeof(struct sctp_status));
socklen_t optLen = (socklen_t) sizeof(struct sctp_status);
if (getsockopt(p5Sock, IPPROTO_SCTP, SCTP_STATUS, &status, &optLen) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After getsockopt errno: %d\n", errno);
return -1;
}
else
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Association ID = %d\n", status.sstat_assoc_id);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Receiver window size = %d\n", status.sstat_rwnd);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF In Streams = %d\n", status.sstat_instrms);
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF Out Streams = %d\n", status.sstat_outstrms);
}
}
}
}
else
{
uint8_t delete_pnfs = 0;
nfapi_vnf_pnf_info_t* pnf = config->pnf_list;
while(pnf != 0)
{
if(FD_ISSET(pnf->p5_sock, &read_fd_set))
{
if(vnf_nr_read_dispatch_message(config, pnf) == 0)
{
if(config->pnf_disconnect_indication != 0)
{
(config->pnf_disconnect_indication)(config, pnf->p5_idx);
}
close(pnf->p5_sock);
pnf->to_delete = 1;
delete_pnfs = 1;
}
}
pnf = pnf->next;
}
if(delete_pnfs)
{
nfapi_vnf_pnf_info_t* pnf = config->pnf_list;
nfapi_vnf_pnf_info_t* prev = 0;
while(pnf != 0)
{
nfapi_vnf_pnf_info_t* curr = pnf;
if(pnf->to_delete == 1)
{
if(prev == 0)
{
config->pnf_list = pnf->next;
}
else
{
prev->next = pnf->next;
}
pnf = pnf->next;
free(curr);
}
else
{
prev = pnf;
pnf = pnf->next;
}
}
}
}
continue;
}
else
{
// timeout
// Should we test for socket closure here every second?
continue;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p5Sock socket's\n");
{
nfapi_vnf_pnf_info_t* curr = config->pnf_list;
while(curr != NULL)
{
if(config->pnf_disconnect_indication)
{
(config->pnf_disconnect_indication)(config, curr->p5_idx);
}
close(curr->p5_sock);
curr = curr->next;
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p5Listen socket\n");
close(p5ListenSock);
return 0;
}
int nfapi_vnf_start(nfapi_vnf_config_t* config)
{

291
nfapi/open-nFAPI/vnf/src/vnf_p7.c
View File

@@ -473,101 +473,6 @@ int vnf_send_p7_msg(vnf_p7_t* vnf_p7, nfapi_vnf_p7_connection_info_t* p7_info, u
return 0;
}
bool vnf_nr_send_p7_msg(vnf_p7_t* vnf_p7, nfapi_nr_p7_message_header_t* header)
{
nfapi_vnf_p7_connection_info_t* p7_connection = vnf_p7_connection_info_list_find(vnf_p7, header->phy_id);
if(p7_connection)
{
int send_result = 0;
uint8_t buffer[1024*1024*3];
header->m_segment_sequence = NFAPI_P7_SET_MSS(0, 0, p7_connection->sequence_number);
int len = nfapi_nr_p7_message_pack(header, buffer, sizeof(buffer), &vnf_p7->_public.codec_config);
//NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() phy_id:%d nfapi_p7_message_pack()=len=%d vnf_p7->_public.segment_size:%u\n", __FUNCTION__, header->phy_id, len, vnf_p7->_public.segment_size);
if(len < 0)
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() failed to pack p7 message phy_id:%d\n", __FUNCTION__, header->phy_id);
return false;
}
if(len > vnf_p7->_public.segment_size)
{
// todo : consider replacing with the sendmmsg call
// todo : worry about blocking writes?
// segmenting the transmit
int msg_body_len = len - NFAPI_NR_P7_HEADER_LENGTH ;
uint32_t seg_body_len = vnf_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 tx_buffer[vnf_p7->_public.segment_size];
NFAPI_TRACE(NFAPI_TRACE_DEBUG, "%s() MORE THAN ONE SEGMENT phy_id:%d nfapi_p7_message_pack()=len=%d vnf_p7->_public.segment_size:%u\n", __FUNCTION__, header->phy_id, len, vnf_p7->_public.segment_size);
for(segment = 0; segment < segment_count; ++segment)
{
uint8_t last = 0;
uint32_t size = vnf_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(&tx_buffer[0], buffer, NFAPI_NR_P7_HEADER_LENGTH);
// set the segment length
tx_buffer[4] = (segment_size & 0xFF000000) >> 24;
tx_buffer[5] = (segment_size & 0xFF0000) >> 16;
tx_buffer[6] = (segment_size & 0xFF00) >> 8;
tx_buffer[7] = (segment_size & 0xFF);
// set the m & segment number
tx_buffer[8] = ((!last) << 7) + segment;
memcpy(&tx_buffer[NFAPI_NR_P7_HEADER_LENGTH], &buffer[0] + offset, size);
offset += size;
if(vnf_p7->_public.checksum_enabled)
{
nfapi_nr_p7_update_checksum(tx_buffer, segment_size);
}
nfapi_nr_p7_update_transmit_timestamp(buffer, calculate_transmit_timestamp(p7_connection->mu, p7_connection->sfn, p7_connection->slot, vnf_p7->slot_start_time_hr));
send_result = vnf_send_p7_msg(vnf_p7, p7_connection, &tx_buffer[0], segment_size);
}
}
else
{
if(vnf_p7->_public.checksum_enabled)
{
nfapi_nr_p7_update_checksum(buffer, len);
}
nfapi_nr_p7_update_transmit_timestamp(buffer, calculate_transmit_timestamp(p7_connection->mu, p7_connection->sfn, p7_connection->slot, vnf_p7->slot_start_time_hr));
// simple case that the message fits in a single segement
send_result = vnf_send_p7_msg(vnf_p7, p7_connection, &buffer[0], len);
}
p7_connection->sequence_number++;
return send_result == 0;
}
else
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() cannot find p7 connection info for phy_id:%d\n", __FUNCTION__, header->phy_id);
return false;
}
}
int vnf_p7_pack_and_send_p7_msg(vnf_p7_t* vnf_p7, nfapi_p7_message_header_t* header)
{
@@ -2220,7 +2125,7 @@ void vnf_dispatch_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7)
}
}
void vnf_nr_dispatch_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7)
void vnf_nr_handle_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7)
{
nfapi_nr_p7_message_header_t header;
@@ -2418,198 +2323,6 @@ void vnf_handle_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7)
}
}
void vnf_nr_handle_p7_message(void *pRecvMsg, int recvMsgLen, vnf_p7_t* vnf_p7)
{
nfapi_nr_p7_message_header_t messageHeader;
// validate the input params
if(pRecvMsg == NULL || recvMsgLen < 4 || vnf_p7 == NULL)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "vnf_handle_p7_message: invalid input params (%p %d %p)\n", pRecvMsg, recvMsgLen, vnf_p7);
return;
}
// unpack the message header
const bool result = vnf_p7->_public.hdr_unpack_func(pRecvMsg, recvMsgLen, &messageHeader, sizeof(nfapi_nr_p7_message_header_t), &vnf_p7->_public.codec_config);
if (!result)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Unpack message header failed, ignoring\n");
return;
}
if(vnf_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 msg:%d len:%d\n", checksum, messageHeader.checksum, messageHeader.message_id, recvMsgLen);
return;
}
}
uint8_t m = NFAPI_P7_GET_MORE(messageHeader.m_segment_sequence);
uint8_t segment_num = NFAPI_P7_GET_SEGMENT(messageHeader.m_segment_sequence);
uint8_t sequence_num = NFAPI_P7_GET_SEQUENCE(messageHeader.m_segment_sequence);
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;
}
//vnf_dispatch_p7_message(&messageHeader, pRecvMsg, recvMsgLen, vnf_p7);
vnf_nr_dispatch_p7_message(pRecvMsg, recvMsgLen, vnf_p7);
}
else
{
nfapi_vnf_p7_connection_info_t* phy = vnf_p7_connection_info_list_find(vnf_p7, messageHeader.phy_id);
if(phy)
{
vnf_p7_rx_message_t* rx_msg = vnf_p7_rx_reassembly_queue_add_segment(vnf_p7, &(phy->reassembly_queue), 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;
uint16_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(phy->reassembly_buffer_size < length)
{
vnf_p7_free(vnf_p7, phy->reassembly_buffer);
phy->reassembly_buffer = 0;
}
if(phy->reassembly_buffer == 0)
{
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Resizing VNF_P7 Reassembly buffer %d->%d\n", phy->reassembly_buffer_size, length);
phy->reassembly_buffer = (uint8_t*)vnf_p7_malloc(vnf_p7, length);
if(phy->reassembly_buffer == 0)
{
NFAPI_TRACE(NFAPI_TRACE_NOTE, "Failed to allocate VNF_P7 reassemby buffer len:%d\n", length);
return;
}
memset(phy->reassembly_buffer, 0, length);
phy->reassembly_buffer_size = length;
}
uint16_t offset = 0;
for(i = 0; i < rx_msg->num_segments_expected; ++i)
{
if(i == 0)
{
memcpy(phy->reassembly_buffer, rx_msg->segments[i].buffer, rx_msg->segments[i].length);
offset += rx_msg->segments[i].length;
}
else
{
memcpy(phy->reassembly_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_dispatch_p7_message(pnf_p7->reassemby_buffer, length, pnf_p7, rx_msg->rx_hr_time);
vnf_nr_dispatch_p7_message(phy->reassembly_buffer, length , vnf_p7);
// delete the structure
vnf_p7_rx_reassembly_queue_remove_msg(vnf_p7, &(phy->reassembly_queue), rx_msg);
}
// see corresponding comment in pnf_nr_handle_p7_message() [same commit]
vnf_p7_rx_reassembly_queue_remove_old_msgs(vnf_p7, &(phy->reassembly_queue), 10000);
}
else
{
NFAPI_TRACE(NFAPI_TRACE_INFO, "Unknown phy id %d\n", messageHeader.phy_id);
}
}
}
int vnf_nr_p7_read_dispatch_message(vnf_p7_t* vnf_p7)
{
int recvfrom_result = 0;
struct sockaddr_in remote_addr;
socklen_t remote_addr_size = sizeof(remote_addr);
do
{
// peek the header
uint8_t header_buffer[NFAPI_NR_P7_HEADER_LENGTH];
recvfrom_result = recvfrom(vnf_p7->socket, 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;
const bool result = vnf_p7->_public.hdr_unpack_func(header_buffer, NFAPI_NR_P7_HEADER_LENGTH, &header, sizeof(header), 0);
if(!result)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "Unpack message header failed, ignoring\n");
return -1;
}
// resize the buffer if we have a large segment
if(header.message_length > vnf_p7->rx_message_buffer_size)
{
NFAPI_TRACE(NFAPI_TRACE_NOTE, "reallocing rx buffer %d\n", header.message_length);
vnf_p7->rx_message_buffer = realloc(vnf_p7->rx_message_buffer, header.message_length);
vnf_p7->rx_message_buffer_size = header.message_length;
}
// read the segment
recvfrom_result = recvfrom(vnf_p7->socket, vnf_p7->rx_message_buffer, header.message_length, MSG_WAITALL | MSG_TRUNC, (struct sockaddr*)&remote_addr, &remote_addr_size);
NFAPI_TRACE(NFAPI_TRACE_DEBUG, "recvfrom_result = %d from %s():%d\n", recvfrom_result, __FUNCTION__, __LINE__);
// todo : how to handle incomplete readfroms, need some sort of buffer/select
if (recvfrom_result > 0)
{
if (recvfrom_result != header.message_length)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "(%d) Received unexpected number of bytes. %d != %d",
__LINE__, recvfrom_result, header.message_length);
break;
}
NFAPI_TRACE(NFAPI_TRACE_DEBUG, "Calling vnf_nr_handle_p7_message from %d\n", __LINE__);
vnf_nr_handle_p7_message(vnf_p7->rx_message_buffer, recvfrom_result, vnf_p7);
return 0;
}
else
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "recvfrom failed %d %d\n", recvfrom_result, errno);
}
}
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);
}
}
}
while(recvfrom_result > 0);
return 0;
}
int vnf_p7_read_dispatch_message(vnf_p7_t* vnf_p7)
{
@@ -2655,7 +2368,7 @@ int vnf_p7_read_dispatch_message(vnf_p7_t* vnf_p7)
__LINE__, recvfrom_result, header.message_length);
break;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Calling vnf_nr_handle_p7_message from %d\n", __LINE__);
NFAPI_TRACE(NFAPI_TRACE_INFO, "Calling vnf_nr_reassemble_p7_message from %d\n", __LINE__);
vnf_handle_p7_message(vnf_p7->rx_message_buffer, recvfrom_result, vnf_p7);
return 0;
}

113
nfapi/open-nFAPI/vnf/src/vnf_p7_interface.c
View File

@@ -30,7 +30,7 @@
#include "common/ran_context.h"
#include "openair1/PHY/defs_gNB.h"
#define FAPI2_IP_DSCP 0
extern RAN_CONTEXT_t RC;
@@ -96,115 +96,6 @@ struct timespec timespec_sub(struct timespec lhs, struct timespec rhs)
// monitor the p7 endpoints and the timing loop and
// send indications to mac
int nfapi_nr_vnf_p7_start(nfapi_vnf_p7_config_t* config)
{
if(config == 0)
return -1;
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s()\n", __FUNCTION__);
vnf_p7_t* vnf_p7 = (vnf_p7_t*)config;
// Create p7 receive udp port
// todo : this needs updating for Ipv6
NFAPI_TRACE(NFAPI_TRACE_INFO, "Initialising VNF P7 port:%u\n", config->port);
// open the UDP socket
if ((vnf_p7->socket = socket(AF_INET, SOCK_DGRAM, 0)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P7 socket errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 socket created...\n");
// configure the UDP socket options
int iptos_value = FAPI2_IP_DSCP << 2;
if (setsockopt(vnf_p7->socket, IPPROTO_IP, IP_TOS, &iptos_value, sizeof(iptos_value)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (IP_TOS) errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 setsockopt succeeded...\n");
// Create the address structure
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(config->port);
addr.sin_addr.s_addr = INADDR_ANY;
// bind to the configured port
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 binding too %s:%d\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
if (bind(vnf_p7->socket, (struct sockaddr *)&addr, sizeof(struct sockaddr_in)) < 0)
//if (sctp_bindx(config->socket, (struct sockaddr *)&addr, sizeof(struct sockaddr_in), 0) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After bind errno: %d\n", errno);
return -1;
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 bind succeeded...\n");
//struct timespec original_pselect_timeout;
struct timespec pselect_timeout;
pselect_timeout.tv_sec = 100;
pselect_timeout.tv_nsec = 0;
struct timespec ref_time;
clock_gettime(CLOCK_MONOTONIC, &ref_time);
while(vnf_p7->terminate == 0)
{
fd_set rfds;
int maxSock = 0;
FD_ZERO(&rfds);
int selectRetval = 0;
// Add the p7 socket
FD_SET(vnf_p7->socket, &rfds);
maxSock = vnf_p7->socket;
selectRetval = pselect(maxSock+1, &rfds, NULL, NULL, &pselect_timeout, NULL);
if(selectRetval == 0)
{
// pselect timed out, continue
}
else if(selectRetval > 0)
{
// have a p7 message
if(FD_ISSET(vnf_p7->socket, &rfds))
{
vnf_nr_p7_read_dispatch_message(vnf_p7);
}
}
else
{
// pselect error
if(selectRetval == -1 && errno == EINTR)
{
// a sigal was received.
}
else
{
//NFAPI_TRACE(NFAPI_TRACE_INFO, "P7 select failed result %d errno %d timeout:%d.%d orginal:%d.%d last_ms:%ld ms:%ld\n", selectRetval, errno, pselect_timeout.tv_sec, pselect_timeout.tv_nsec, pselect_timeout.tv_sec, pselect_timeout.tv_nsec, last_millisecond, millisecond);
// should we exit now?
if (selectRetval == -1 && errno == 22) // invalid argument??? not sure about timeout duration
{
usleep(100000);
}
}
}
}
NFAPI_TRACE(NFAPI_TRACE_INFO, "Closing p7 socket\n");
close(vnf_p7->socket);
NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() returning\n", __FUNCTION__);
return 0;
}
int nfapi_vnf_p7_start(nfapi_vnf_p7_config_t* config)
@@ -231,7 +122,7 @@ int nfapi_vnf_p7_start(nfapi_vnf_p7_config_t* config)
NFAPI_TRACE(NFAPI_TRACE_INFO, "VNF P7 socket created...\n");
// configure the UDP socket options
int iptos_value = FAPI2_IP_DSCP << 2;
int iptos_value = 0;
if (setsockopt(vnf_p7->socket, IPPROTO_IP, IP_TOS, &iptos_value, sizeof(iptos_value)) < 0)
{
NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt (IP_TOS) errno: %d\n", errno);