From bfaec58ca50e770c2c3bc0b780bbcd30e8459db7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=BAben=20Soares=20Silva?= Date: Mon, 17 Mar 2025 12:58:40 +0000 Subject: [PATCH] Separate PNF socket send functions into separate socket library --- CMakeLists.txt | 4 + nfapi/CMakeLists.txt | 1 + nfapi/oai_integration/nfapi_pnf.c | 39 +- nfapi/oai_integration/socket/CMakeLists.txt | 13 + .../socket/include/socket_common.h | 39 + .../socket/include/socket_pnf.h | 39 + nfapi/oai_integration/socket/socket_common.c | 58 ++ nfapi/oai_integration/socket/socket_pnf.c | 867 ++++++++++++++++++ nfapi/open-nFAPI/fapi/CMakeLists.txt | 2 + .../nfapi/public_inc/nfapi_interface.h | 3 +- nfapi/open-nFAPI/nfapi/src/nfapi_p7.c | 3 + nfapi/open-nFAPI/pnf/inc/pnf.h | 3 +- nfapi/open-nFAPI/pnf/inc/pnf_p7.h | 8 +- .../pnf/public_inc/nfapi_pnf_interface.h | 3 +- nfapi/open-nFAPI/pnf/src/pnf.c | 263 +----- nfapi/open-nFAPI/pnf/src/pnf_interface.c | 30 - nfapi/open-nFAPI/pnf/src/pnf_p7.c | 448 +-------- nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c | 13 - 18 files changed, 1089 insertions(+), 747 deletions(-) create mode 100644 nfapi/oai_integration/socket/CMakeLists.txt create mode 100644 nfapi/oai_integration/socket/include/socket_common.h create mode 100644 nfapi/oai_integration/socket/include/socket_pnf.h create mode 100644 nfapi/oai_integration/socket/socket_common.c create mode 100644 nfapi/oai_integration/socket/socket_pnf.c diff --git a/CMakeLists.txt b/CMakeLists.txt index cc30b7a5bd..4dc8c9ce7a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -715,6 +715,7 @@ add_library(NFAPI_LIB ${NFAPI_SRC}) target_link_libraries(NFAPI_LIB PUBLIC nfapi_common) target_link_libraries(NFAPI_LIB PUBLIC nr_fapi_p5 nr_fapi_p7) +target_compile_definitions(NFAPI_LIB PRIVATE ENABLE_SOCKET) include_directories(${NFAPI_DIR}/nfapi/public_inc) include_directories(${NFAPI_DIR}/nfapi/inc) @@ -755,6 +756,7 @@ set(NFAPI_USER_SRC ${NFAPI_USER_DIR}/gnb_ind_vars.c ) add_library(NFAPI_USER_LIB ${NFAPI_USER_SRC}) +target_compile_definitions(NFAPI_USER_LIB PRIVATE ENABLE_SOCKET) target_link_libraries(NFAPI_USER_LIB PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs UTIL) target_link_libraries(NFAPI_USER_LIB PRIVATE nr_fapi_p7) if(OAI_AERIAL) @@ -1839,6 +1841,8 @@ target_link_libraries(nr-softmodem PRIVATE pthread m CONFIG_LIB rt sctp) target_link_libraries(nr-softmodem PRIVATE ${T_LIB}) target_link_libraries(nr-softmodem PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs) target_link_libraries(nr-softmodem PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs) + +target_link_libraries(nr-softmodem PUBLIC nfapi_socket_lib) add_boolean_option(OAI_AERIAL OFF "Activate OAI's AERIAL driver" OFF) if (OAI_AERIAL) target_compile_definitions(nr-softmodem PUBLIC ENABLE_AERIAL) diff --git a/nfapi/CMakeLists.txt b/nfapi/CMakeLists.txt index d6c5b0e92a..e79905b07c 100644 --- a/nfapi/CMakeLists.txt +++ b/nfapi/CMakeLists.txt @@ -1,3 +1,4 @@ add_subdirectory(oai_integration/aerial) +add_subdirectory(oai_integration/socket) add_subdirectory(open-nFAPI) add_subdirectory(tests) diff --git a/nfapi/oai_integration/nfapi_pnf.c b/nfapi/oai_integration/nfapi_pnf.c index e17690da2d..88b82ad480 100644 --- a/nfapi/oai_integration/nfapi_pnf.c +++ b/nfapi/oai_integration/nfapi_pnf.c @@ -63,6 +63,10 @@ #include "executables/lte-softmodem.h" #include "nfapi/open-nFAPI/pnf/inc/pnf_p7.h" +#ifdef ENABLE_SOCKET +#include +#endif + #define NUM_P5_PHY 2 #define _GNU_SOURCE @@ -186,7 +190,7 @@ typedef struct { } pnf_phy_user_data_t; static pnf_info pnf; -static pthread_t pnf_start_pthread; +static pthread_t pnf_p5_init_and_receive_pthread; int nfapitooai_level(int nfapi_level) { switch(nfapi_level) { @@ -220,12 +224,7 @@ void *pnf_p7_thread_start(void *ptr) { return 0; } -void *pnf_nr_p7_thread_start(void *ptr) { - NFAPI_TRACE(NFAPI_TRACE_INFO, "[NR_PNF] NR P7 THREAD %s\n", __FUNCTION__); - nfapi_pnf_p7_config_t *config = (nfapi_pnf_p7_config_t *)ptr; - nfapi_nr_pnf_p7_start(config); - return 0; -} + int pnf_nr_param_request(nfapi_pnf_config_t *config, nfapi_nr_pnf_param_request_t *req) { @@ -1752,6 +1751,8 @@ int nr_start_request(nfapi_pnf_config_t *config, nfapi_pnf_phy_config_t *phy, nf p7_config->codec_config.pack_p7_vendor_extension = &pnf_nr_phy_pack_p7_vendor_extension; p7_config->codec_config.unpack_vendor_extension_tlv = &pnf_nr_phy_unpack_vendor_extension_tlv; p7_config->codec_config.pack_vendor_extension_tlv = &pnf_nr_phy_pack_vendor_extention_tlv; + +#ifdef ENABLE_SOCKET p7_config->unpack_func = &nfapi_nr_p7_message_unpack; p7_config->hdr_unpack_func = &nfapi_nr_p7_message_header_unpack; p7_config->pack_func = &nfapi_nr_p7_message_pack; @@ -1759,6 +1760,8 @@ int nr_start_request(nfapi_pnf_config_t *config, nfapi_pnf_phy_config_t *phy, nf NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] Creating P7 thread %s\n", __FUNCTION__); pthread_t p7_thread; threadCreate(&p7_thread, &pnf_nr_p7_thread_start, p7_config, "pnf_p7_thread", -1, OAI_PRIORITY_RT); +#endif + NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] Calling l1_north_init_eNB() %s\n", __FUNCTION__); l1_north_init_gNB(); NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] HACK - Set p7_config global ready for subframe ind%s\n", __FUNCTION__); @@ -2141,15 +2144,6 @@ void *pnf_start_thread(void *ptr) { return (void *)0; } -void *pnf_nr_start_thread(void *ptr) { - NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] IN PNF NFAPI start thread %s\n", __FUNCTION__); - nfapi_pnf_config_t *config = (nfapi_pnf_config_t *)ptr; - struct sched_param sp; - sp.sched_priority = 20; - pthread_setschedparam(pthread_self(),SCHED_FIFO,&sp); - nfapi_nr_pnf_start(config); - return (void *)0; -} void configure_nr_nfapi_pnf(char *vnf_ip_addr, int vnf_p5_port, char *pnf_ip_addr, int pnf_p7_port, int vnf_p7_port) { @@ -2197,13 +2191,18 @@ void configure_nr_nfapi_pnf(char *vnf_ip_addr, int vnf_p5_port, char *pnf_ip_add config->deallocate_p4_p5_vendor_ext = &pnf_nr_sim_deallocate_p4_p5_vendor_ext; config->codec_config.unpack_p4_p5_vendor_extension = &pnf_nr_sim_unpack_p4_p5_vendor_extension; config->codec_config.pack_p4_p5_vendor_extension = &pnf_nr_sim_pack_p4_p5_vendor_extension; + + +#ifdef ENABLE_SOCKET + NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] Creating PNF NFAPI start thread %s\n", __FUNCTION__); 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 = &pnf_nr_send_p5_message; NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] Creating PNF NFAPI start thread %s\n", __FUNCTION__); - pthread_create(&pnf_start_pthread, NULL, &pnf_nr_start_thread, config); - pthread_setname_np(pnf_start_pthread, "NFAPI_PNF"); + pthread_create(&pnf_p5_init_and_receive_pthread, NULL, &pnf_start_p5_thread, config); + pthread_setname_np(pnf_p5_init_and_receive_pthread, "NFAPI_PNF"); +#endif } void configure_nfapi_pnf(char *vnf_ip_addr, int vnf_p5_port, char *pnf_ip_addr, int pnf_p7_port, int vnf_p7_port) { @@ -2253,8 +2252,8 @@ void configure_nfapi_pnf(char *vnf_ip_addr, int vnf_p5_port, char *pnf_ip_addr, config->codec_config.unpack_p4_p5_vendor_extension = &pnf_sim_unpack_p4_p5_vendor_extension; config->codec_config.pack_p4_p5_vendor_extension = &pnf_sim_pack_p4_p5_vendor_extension; NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] Creating PNF NFAPI start thread %s\n", __FUNCTION__); - pthread_create(&pnf_start_pthread, NULL, &pnf_start_thread, config); - pthread_setname_np(pnf_start_pthread, "NFAPI_PNF"); + pthread_create(&pnf_p5_init_and_receive_pthread, NULL, &pnf_start_thread, config); + pthread_setname_np(pnf_p5_init_and_receive_pthread, "NFAPI_PNF"); } void oai_subframe_ind(uint16_t sfn, uint16_t sf) { diff --git a/nfapi/oai_integration/socket/CMakeLists.txt b/nfapi/oai_integration/socket/CMakeLists.txt new file mode 100644 index 0000000000..edf5457cea --- /dev/null +++ b/nfapi/oai_integration/socket/CMakeLists.txt @@ -0,0 +1,13 @@ +add_library(nfapi_socket_lib + socket_pnf.c + socket_common.c +) + +target_include_directories(nfapi_socket_lib PRIVATE include/) + +target_compile_definitions(nfapi_socket_lib PUBLIC ENABLE_SOCKET) +target_link_libraries(nfapi_socket_lib PRIVATE NFAPI_LIB NFAPI_USER_LIB) +target_link_libraries(nfapi_socket_lib PRIVATE LOG) +add_dependencies(nfapi_socket_lib generate_T) +target_link_libraries(nfapi_socket_lib PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs) +target_link_libraries(nfapi_socket_lib PRIVATE nr_fapi_p5 nr_fapi_p7) diff --git a/nfapi/oai_integration/socket/include/socket_common.h b/nfapi/oai_integration/socket/include/socket_common.h new file mode 100644 index 0000000000..2456740ec4 --- /dev/null +++ b/nfapi/oai_integration/socket/include/socket_common.h @@ -0,0 +1,39 @@ +/* + * 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_COMMON_H +#define SOCKET_COMMON_H + +#include +#include +#include +#include +#include +#include + +int socket_send_p5_msg(const int sctp, + const int socket, + const void* remote_addr, + const void* msg, + const uint32_t len, + const uint16_t stream); +int socket_send_p7_msg(const int socket, const void* remote_addr, const void* msg, const uint32_t len); + +#endif // SOCKET_COMMON_H diff --git a/nfapi/oai_integration/socket/include/socket_pnf.h b/nfapi/oai_integration/socket/include/socket_pnf.h new file mode 100644 index 0000000000..ccb3eaea3a --- /dev/null +++ b/nfapi/oai_integration/socket/include/socket_pnf.h @@ -0,0 +1,39 @@ +/* + * 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_PNF_H +#define SOCKET_PNF_H +#include "socket_common.h" +#include +#include +#include +#include "pnf.h" +#include "pnf_p7.h" +#include "nr_fapi_p5.h" +#include "nr_nfapi_p7.h" + +int pnf_connect_socket(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); +bool pnf_nr_send_p7_message(pnf_p7_t* pnf_p7, nfapi_nr_p7_message_header_t* header, uint32_t msg_len); +void *pnf_start_p5_thread(void *ptr); +void *pnf_nr_p7_thread_start(void *ptr); +#endif // SOCKET_PNF_H diff --git a/nfapi/oai_integration/socket/socket_common.c b/nfapi/oai_integration/socket/socket_common.c new file mode 100644 index 0000000000..5ec9a13ac6 --- /dev/null +++ b/nfapi/oai_integration/socket/socket_common.c @@ -0,0 +1,58 @@ +/* + * 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_common.h" +#include "LOG/log.h" + +int socket_send_p5_msg(const int sctp, + const int socket, + const void* remote_addr, + const void* msg, + const uint32_t len, + const uint16_t stream) +{ + if (sctp == 1) { + struct sockaddr* to = (struct sockaddr*)remote_addr; + const int retval = sctp_sendmsg(socket, msg, len, to, 0, 42, 0, stream, 0, 0); + if (retval < 0) { + LOG_E(NR_PHY, "sctp_sendmsg failed errno: %d\n", errno); + return -1; + } + } else { + if (send(socket, msg, len, 0) != len) { + LOG_E(NR_PHY, "write failed errno: %d\n", errno); + return -1; + } + } + + return len; +} + +int socket_send_p7_msg(const int socket, const void* remote_addr, const void* msg, const uint32_t len) +{ + const struct sockaddr* to = (struct sockaddr*)remote_addr; + const long sendto_result = sendto(socket, msg, len, 0, to, sizeof(struct sockaddr_in)); + if (sendto_result != len) { + LOG_E(NR_PHY, "%s() sendto_result %ld %d\n", __FUNCTION__, sendto_result, errno); + return -1; + } + + return 0; +} diff --git a/nfapi/oai_integration/socket/socket_pnf.c b/nfapi/oai_integration/socket/socket_pnf.c new file mode 100644 index 0000000000..fe4999c998 --- /dev/null +++ b/nfapi/oai_integration/socket/socket_pnf.c @@ -0,0 +1,867 @@ +/* + * 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_pnf.h" +#include "nfapi.h" + +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, "p5_message_pack failed (%d)\n", packed_len); + return false; + } + + return socket_send_p5_msg(pnf->sctp, pnf->p5_sock, NULL, pnf->tx_message_buffer, packed_len, 0) == packed_len; +} + +static int send_p7_msg(pnf_p7_t *pnf_p7, uint8_t *msg, uint32_t len) +{ + // todo : consider how to do this only once + struct sockaddr_in remote_addr = {.sin_family = AF_INET, .sin_port = htons(pnf_p7->_public.remote_p7_port)}; + + if (inet_aton(pnf_p7->_public.remote_p7_addr, &remote_addr.sin_addr) == -1) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "inet_aton failed %d\n", errno); + return -1; + } + + socklen_t remote_addr_len = sizeof(struct sockaddr_in); + + int sendto_result = socket_send_p7_msg((int)pnf_p7->p7_sock, &remote_addr, msg, len); + if (sendto_result < 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; + } + + if (sendto_result != 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "%s sendto failed to send the entire message %d %d\n", __FUNCTION__, sendto_result, len); + } + 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_NR_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 , update and push m_segment_sequence + uint8_t *buf_ptr = &buffer[4]; + uint8_t *buf_ptr_end = &buffer[10]; + push32(segment_size, (&buf_ptr), buf_ptr_end); + header->m_segment_sequence = NFAPI_NR_P7_SET_MSS((!last), segment, pnf_p7->sequence_number); + push16(header->m_segment_sequence, (&buf_ptr), buf_ptr_end); + 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); + } + + send_p7_msg(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 + send_p7_msg(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_start_p5_thread(void *ptr) +{ + NFAPI_TRACE(NFAPI_TRACE_INFO, "[PNF] IN PNF NFAPI start thread %s\n", __FUNCTION__); + nfapi_pnf_config_t *config = (nfapi_pnf_config_t *)ptr; + struct sched_param sp; + sp.sched_priority = 20; + pthread_setschedparam(pthread_self(), SCHED_FIFO, &sp); + + // Verify that config is not null + if (config == 0) + return (void *)-1; + + NFAPI_TRACE(NFAPI_TRACE_INFO, "%s\n", __FUNCTION__); + + pnf_t *_this = (pnf_t *)(config); + + while (_this->terminate == 0) { + int connect_result = pnf_connect_socket(_this); + + if (connect_result > 0) { + pnf_nr_message_pump(_this); + } else if (connect_result < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "%s() Error connecting to P5 socket\n", __FUNCTION__); + return 0; + } + + sleep(1); + } + NFAPI_TRACE(NFAPI_TRACE_INFO, "%s() terminate=1 - EXITTING............\n", __FUNCTION__); + + return 0; +} + +int pnf_connect_socket(pnf_t *pnf) +{ + uint8_t socketConnected = 0; + + if (pnf->_public.vnf_ip_addr == 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "vnfIpAddress is null\n"); + return -1; + } + + NFAPI_TRACE(NFAPI_TRACE_INFO, "Starting P5 PNF connection to VNF at %s:%u\n", pnf->_public.vnf_ip_addr, pnf->_public.vnf_p5_port); + + // todo split the vnf address list. currently only supporting 1 + + struct addrinfo hints = {0}, *servinfo; + + hints.ai_socktype = SOCK_STREAM; // For SCTP we are only interested in SOCK_STREAM + // todo : allow the client to restrict IPV4 or IPV6 + // todo : randomize which address to connect to? + + char port_str[8]; + snprintf(port_str, sizeof(port_str), "%d", pnf->_public.vnf_p5_port); + if (getaddrinfo(pnf->_public.vnf_ip_addr, port_str, &hints, &servinfo) != 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "Failed to get host (%s) addr info h_errno:%d \n", pnf->_public.vnf_ip_addr, h_errno); + return -1; + } + + struct addrinfo *p = servinfo; + int connected = 0; + + while (p != NULL && connected == 0) { +#ifdef NFAPI_TRACE_ENABLED + char *family = "Unknown"; + char *address = "Unknown"; + char _addr[128]; + + if (p->ai_family == AF_INET6) { + family = "IPV6"; + struct sockaddr_in6 *addr = (struct sockaddr_in6 *)p->ai_addr; + inet_ntop(AF_INET6, &addr->sin6_addr, _addr, sizeof(_addr)); + address = &_addr[0]; + } else if (p->ai_family == AF_INET) { + family = "IPV4"; + struct sockaddr_in *addr = (struct sockaddr_in *)p->ai_addr; + address = inet_ntoa(addr->sin_addr); + } + + NFAPI_TRACE(NFAPI_TRACE_NOTE, "Host address info %d Family:%s Address:%s\n", i++, family, address); +#endif + + if (pnf->sctp) { + // open the SCTP socket + if ((pnf->p5_sock = socket(p->ai_family, SOCK_STREAM, IPPROTO_SCTP)) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P5 socket errno: %d\n", errno); + freeaddrinfo(servinfo); + return -1; + } + int noDelay; + struct sctp_initmsg initMsg = {0}; + + // configure the socket options + NFAPI_TRACE(NFAPI_TRACE_NOTE, "PNF Setting the SCTP_INITMSG\n"); + 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(pnf->p5_sock, IPPROTO_SCTP, SCTP_INITMSG, &initMsg, sizeof(initMsg)) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt errno: %d\n", errno); + freeaddrinfo(servinfo); + return -1; + } + noDelay = 1; + if (setsockopt(pnf->p5_sock, IPPROTO_SCTP, SCTP_NODELAY, &noDelay, sizeof(noDelay)) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt errno: %d\n", errno); + freeaddrinfo(servinfo); + return -1; + } + + struct sctp_event_subscribe events = {0}; + events.sctp_data_io_event = 1; + + if (setsockopt(pnf->p5_sock, SOL_SCTP, SCTP_EVENTS, (const void *)&events, sizeof(events)) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After setsockopt errno: %d\n", errno); + freeaddrinfo(servinfo); + return -1; + } + } else { + // Create an IP socket point + if ((pnf->p5_sock = socket(p->ai_family, SOCK_STREAM, IPPROTO_IP)) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After P5 socket errno: %d\n", errno); + freeaddrinfo(servinfo); + return -1; + } + } + + NFAPI_TRACE(NFAPI_TRACE_INFO, "P5 socket created...\n"); + + if (connect(pnf->p5_sock, p->ai_addr, p->ai_addrlen) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, + "After connect (address:%s port:%d) errno: %d\n", + pnf->_public.vnf_ip_addr, + pnf->_public.vnf_p5_port, + errno); + + if (errno == EINVAL) { + freeaddrinfo(servinfo); + return -1; + } + if (pnf->terminate != 0) { + freeaddrinfo(servinfo); + return 0; + } + close(pnf->p5_sock); + sleep(1); + } else { + NFAPI_TRACE(NFAPI_TRACE_INFO, "connect succeeded...\n"); + + connected = 1; + } + + p = p->ai_next; + } + + freeaddrinfo(servinfo); + + // If we have failed to connect return 0 and it is retry + if (connected == 0) + return 0; + + NFAPI_TRACE(NFAPI_TRACE_NOTE, "After connect loop\n"); + if (pnf->sctp) { + socklen_t optLen; + struct sctp_status status = {0}; + + // check the connection status + optLen = (socklen_t)sizeof(struct sctp_status); + if (getsockopt(pnf->p5_sock, 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, "Association ID = %d\n", status.sstat_assoc_id); + NFAPI_TRACE(NFAPI_TRACE_INFO, "Receiver window size = %d\n", status.sstat_rwnd); + NFAPI_TRACE(NFAPI_TRACE_INFO, "In Streams = %d\n", status.sstat_instrms); + NFAPI_TRACE(NFAPI_TRACE_INFO, "Out Streams = %d\n", status.sstat_outstrms); + + socketConnected = 1; + } + } else { + int error; + socklen_t len = sizeof(error); + + if (getsockopt(pnf->p5_sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) { + NFAPI_TRACE(NFAPI_TRACE_ERROR, "After getsockopt errno: %d\n", errno); + return -1; + } else { + // If error is zero, the socket is connected + if (error == 0) { + socketConnected = 1; + } + } + } + + NFAPI_TRACE(NFAPI_TRACE_NOTE, "Socket %s\n", socketConnected ? "CONNECTED" : "NOT_CONNECTED"); + return socketConnected; +} + +static 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 = {0}; + + { + 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; + + 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) { + int tmp = errno; + NFAPI_TRACE(NFAPI_TRACE_INFO, "Failed to read sctp message size error %s errno:%d\n", strerror(tmp), tmp); + } else { +#if 0 + // print the received message + printf("\n MESSAGE RECEIVED: \n"); + for(int i=0; ip5_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 { + 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; +} + +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_TRACE(NFAPI_TRACE_ERROR, "close(sctpSock) failed errno: %d\n", errno); + } + + return 0; +} + +static void pnf_nr_reassemble_nfapi_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_handle_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_handle_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); +} + +static 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_reassemble_nfapi_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_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 = 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); + return -1; + } + + struct sockaddr_in addr = {0}; + 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; +} + +static 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; +} + +void *pnf_nr_p7_thread_start(void *ptr) +{ + NFAPI_TRACE(NFAPI_TRACE_INFO, "[NR_PNF] NR P7 THREAD %s\n", __FUNCTION__); + nfapi_pnf_p7_config_t *config = (nfapi_pnf_p7_config_t *)ptr; + nfapi_nr_pnf_p7_start(config); + return 0; +} diff --git a/nfapi/open-nFAPI/fapi/CMakeLists.txt b/nfapi/open-nFAPI/fapi/CMakeLists.txt index b332ef6a8e..a97f5f6e6f 100644 --- a/nfapi/open-nFAPI/fapi/CMakeLists.txt +++ b/nfapi/open-nFAPI/fapi/CMakeLists.txt @@ -22,3 +22,5 @@ if (OAI_AERIAL) target_compile_definitions(nr_fapi_p5 PRIVATE ENABLE_AERIAL) target_compile_definitions(nr_fapi_p7 PRIVATE ENABLE_AERIAL) endif () + +target_compile_definitions(nr_fapi_common PRIVATE ENABLE_SOCKET) diff --git a/nfapi/open-nFAPI/nfapi/public_inc/nfapi_interface.h b/nfapi/open-nFAPI/nfapi/public_inc/nfapi_interface.h index f830e904e2..3cfd0591fc 100644 --- a/nfapi/open-nFAPI/nfapi/public_inc/nfapi_interface.h +++ b/nfapi/open-nFAPI/nfapi/public_inc/nfapi_interface.h @@ -115,7 +115,8 @@ typedef struct { #define NFAPI_P7_GET_MORE(_mss) ( ((_mss) & 0x8000) >> 15 ) #define NFAPI_P7_GET_SEGMENT(_mss) ( ((_mss) & 0x7F00) >> 8 ) #define NFAPI_P7_GET_SEQUENCE(_mss) ( (_mss) & 0x00FF ) -#define NFAPI_P7_SET_MSS(_more, _segm, _sequ) ( (((_more) & 0x1) << 7) | (((_segm) & 0x7) << 4) | ((_sequ) & 0xF) ) +#define NFAPI_NR_P7_SET_MSS(_more, _segm, _sequ) ( (((_more) & 0x1) << 15) | (((_segm) & 0x7F) << 8) | ((_sequ) & 0xFF) ) +#define NFAPI_P7_SET_MSS(_more, _segm, _sequ) ( (((_more) & 0x1) << 7) | (((_segm) & 0x7F) << 4) | ((_sequ) & 0xF) ) #define NFAPI_TAG_LENGTH_PACKED_LEN 4 diff --git a/nfapi/open-nFAPI/nfapi/src/nfapi_p7.c b/nfapi/open-nFAPI/nfapi/src/nfapi_p7.c index c60ed27d07..f7c55cd2e0 100644 --- a/nfapi/open-nFAPI/nfapi/src/nfapi_p7.c +++ b/nfapi/open-nFAPI/nfapi/src/nfapi_p7.c @@ -6203,7 +6203,10 @@ int nfapi_p7_message_unpack(void *pMessageBuf, bool peek_nr_nfapi_p7_sfn_slot(void *pMessageBuf, uint32_t messageBufLen, uint16_t *SFN, uint16_t *Slot) { +#ifdef ENABLE_SOCKET + // Socket transport will always use nFAPI encoding uint8_t *pReadPackedMessage = &((uint8_t *)pMessageBuf)[NFAPI_NR_P7_HEADER_LENGTH]; +#endif uint8_t *end = (uint8_t*)pMessageBuf + messageBufLen; if(!(pull16(&pReadPackedMessage, SFN, end) && pull16(&pReadPackedMessage, Slot, end))) { NFAPI_TRACE(NFAPI_TRACE_ERROR, "Failed to peek SFN.Slot\n"); diff --git a/nfapi/open-nFAPI/pnf/inc/pnf.h b/nfapi/open-nFAPI/pnf/inc/pnf.h index 386cb3cf6a..968359dfee 100644 --- a/nfapi/open-nFAPI/pnf/inc/pnf.h +++ b/nfapi/open-nFAPI/pnf/inc/pnf.h @@ -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); diff --git a/nfapi/open-nFAPI/pnf/inc/pnf_p7.h b/nfapi/open-nFAPI/pnf/inc/pnf_p7.h index 8a603ecdb3..49868bde8d 100644 --- a/nfapi/open-nFAPI/pnf/inc/pnf_p7.h +++ b/nfapi/open-nFAPI/pnf/inc/pnf_p7.h @@ -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_ */ diff --git a/nfapi/open-nFAPI/pnf/public_inc/nfapi_pnf_interface.h b/nfapi/open-nFAPI/pnf/public_inc/nfapi_pnf_interface.h index 04abf6309d..957e032f69 100644 --- a/nfapi/open-nFAPI/pnf/public_inc/nfapi_pnf_interface.h +++ b/nfapi/open-nFAPI/pnf/public_inc/nfapi_pnf_interface.h @@ -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 diff --git a/nfapi/open-nFAPI/pnf/src/pnf.c b/nfapi/open-nFAPI/pnf/src/pnf.c index e6fb69bd71..fd572cf77f 100644 --- a/nfapi/open-nFAPI/pnf/src/pnf.c +++ b/nfapi/open-nFAPI/pnf/src/pnf.c @@ -30,6 +30,10 @@ #include "nfapi_nr_interface_scf.h" #include "pnf.h" +#include + +#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; ip5_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; ip5_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) { diff --git a/nfapi/open-nFAPI/pnf/src/pnf_interface.c b/nfapi/open-nFAPI/pnf/src/pnf_interface.c index 78ea46bbf6..d80ab168fa 100644 --- a/nfapi/open-nFAPI/pnf/src/pnf_interface.c +++ b/nfapi/open-nFAPI/pnf/src/pnf_interface.c @@ -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 diff --git a/nfapi/open-nFAPI/pnf/src/pnf_p7.c b/nfapi/open-nFAPI/pnf/src/pnf_p7.c index 95fff37bc9..e2b6d253d2 100644 --- a/nfapi/open-nFAPI/pnf/src/pnf_p7.c +++ b/nfapi/open-nFAPI/pnf/src/pnf_p7.c @@ -32,7 +32,6 @@ #include "common/ran_context.h" #include -#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; -} diff --git a/nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c b/nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c index 06d268a875..b021e67fc9 100644 --- a/nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c +++ b/nfapi/open-nFAPI/pnf/src/pnf_p7_interface.c @@ -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)