SPGWU thread scheduler and priority in config file (see boost asio later)

etc/spgw_u.conf

@@ -18,12 +18,34 @@
 # For more information about the OpenAirInterface (OAI) Software Alliance:
 #      contact@openairinterface.org
 ################################################################################
-SPGW-U = 
+SPGW-U =
 {
     INSTANCE                       = @INSTANCE@;            # 0 is the default
     PID_DIRECTORY                  = "@PID_DIRECTORY@";     # /var/run is the default
 
-    INTERFACES : 
+    ITTI :
+    {
+        SCHED_PARAMS :
+        {
+            CPU_ID       = 1;
+            SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+            SCHED_PRIORITY = 85;
+        };
+        S1U_SCHED_PARAMS :
+        {
+            CPU_ID       = 1;
+            SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+            SCHED_PRIORITY = 84;
+        };
+        SX_SCHED_PARAMS :
+        {
+            CPU_ID       = 1;
+            SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+            SCHED_PRIORITY = 84;
+        };
+    };
+
+    INTERFACES :
     {
         S1U_S12_S4_UP :
         {
@@ -31,7 +53,12 @@ SPGW-U =
             INTERFACE_NAME         = "@SGW_INTERFACE_NAME_FOR_S1U_S12_S4_UP@";  # STRING, interface name, YOUR NETWORK CONFIG HERE
             IPV4_ADDRESS           = "read";                                # STRING, CIDR or read to let app read interface configured IP address
             PORT                   = @SGW_UDP_PORT_FOR_S1U_S12_S4_UP@;          # Default is 2152
-            CPU_ID_THREAD_LOOP_READ = 1;
+            SCHED_PARAMS :
+            {
+                CPU_ID       = 2;
+                SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+                SCHED_PRIORITY = 98;
+            };
         };
         SX :
         {
@@ -39,25 +66,36 @@ SPGW-U =
             INTERFACE_NAME         = "@SGW_INTERFACE_NAME_FOR_SX@"; # STRING, interface name
             IPV4_ADDRESS           = "read";                                # STRING, CIDR or read to let app read interface configured IP address
             PORT                   = 8805;                                  # Default is 8805
+            SCHED_PARAMS :
+            {
+                CPU_ID       = 1;
+                SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+                SCHED_PRIORITY = 95;
+            };
         };
         SGI :
         {
            # No config to set, the software will set the SGi interface to the interface used for the default route.
             INTERFACE_NAME         = "@SGW_INTERFACE_NAME_FOR_SGI@"; # STRING, interface name or "default_gateway"
             IPV4_ADDRESS           = "read";                         # STRING, CIDR or read to let app read interface configured IP address
-            CPU_ID_THREAD_LOOP_READ = 3;
+            SCHED_PARAMS :
+            {
+                CPU_ID       = 3;
+                SCHED_POLICY = "SCHED_FIFO"; # Values in { SCHED_OTHER, SCHED_IDLE, SCHED_BATCH, SCHED_FIFO, SCHED_RR }
+                SCHED_PRIORITY = 98;
+            };
         };
     };
 
     PDN_NETWORK_LIST  = (
-                      {NETWORK_IPV4 = "12.1.1.0/24"; NETWORK_IPV6 = "2001:1:2::0/64"; SNAT = "yes";}, 
+                      {NETWORK_IPV4 = "12.1.1.0/24"; NETWORK_IPV6 = "2001:1:2::0/64"; SNAT = "yes";},
                       {NETWORK_IPV4 = "12.1.2.0/24"; SNAT = "no";},
                       {NETWORK_IPV4 = "192.169.0.0/24"; SNAT = "no";},
                       {NETWORK_IPV4 = "192.170.0.0/24"; SNAT = "no";},
                       {NETWORK_IPV4 = "192.171.0.0/24"; SNAT = "no";}
                     );
 
-    SPGW-C_LIST = ( 
+    SPGW-C_LIST = (
          {IPV4_ADDRESS="192.168.160.100" ;}
     );
 };

src/common/thread_sched.h

+/*
+ * 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
+ */
+
+/*! \file thread_sched.h
+  \brief
+  \company Eurecom
+  \email: lionel.gauthier@eurecom.fr
+*/
+
+#ifndef FILE_THREAD_SCHED_SEEN
+#define FILE_THREAD_SCHED_SEEN
+
+#include <sched.h>
+
+typedef struct thread_sched_params_s {
+  int             cpu_id;
+  int             sched_policy;
+  int             sched_priority;
+} thread_sched_params_t;
+#endif /* FILE_COMMON_ROOT_TYPES_SEEN */

src/gtpv1u/gtpv1u.cpp

@@ -55,23 +55,23 @@ static std::string string_to_hex(const std::string& input)
     return output;
 }
 //------------------------------------------------------------------------------
-void udp_server::udp_read_loop(int cpu)
+void udp_server::udp_read_loop(const thread_sched_params_t& thread_sched_params)
 {
   socklen_t                               r_endpoint_addr_len = 0;
   struct sockaddr_storage                 r_endpoint = {};
   size_t                                  bytes_received = 0;
 
-  if (cpu) {
+  if (thread_sched_params.cpu_id >= 0) {
     cpu_set_t cpuset;
-    CPU_SET(cpu,&cpuset);
+    CPU_SET(thread_sched_params.cpu_id ,&cpuset);
     pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
   }
-  int policy;
+
   struct sched_param sparam;
   memset(&sparam, 0, sizeof(sparam));
-  sparam.sched_priority = sched_get_priority_max(SCHED_FIFO);
-  policy = SCHED_FIFO ;
-  pthread_setschedparam(pthread_self(), policy, &sparam);
+  sparam.sched_priority = thread_sched_params.sched_priority;
+  pthread_setschedparam(pthread_self(), thread_sched_params.sched_policy, &sparam);
+
 
   while (1) {
     r_endpoint_addr_len = sizeof(struct sockaddr_storage);
@@ -169,11 +169,11 @@ int udp_server::create_socket (char * address, const uint16_t port_num)
   }
 }
 //------------------------------------------------------------------------------
-void udp_server::start_receive(gtpu_l4_stack * gtp_stack, const int cpu)
+void udp_server::start_receive(gtpu_l4_stack * gtp_stack, const thread_sched_params_t& thread_sched_params)
 {
   app_ = gtp_stack;
   Logger::udp().trace( "udp_server::start_receive");
-  thread_ = thread(&udp_server::udp_read_loop,this, cpu);
+  thread_ = thread(&udp_server::udp_read_loop,this, thread_sched_params);
   thread_.detach();
 }
 
@@ -196,7 +196,7 @@ void udp_server::start_receive(gtpu_l4_stack * gtp_stack, const int cpu)
 //}
 
 //------------------------------------------------------------------------------
-gtpu_l4_stack::gtpu_l4_stack(const struct in_addr& address, const uint16_t port_num, const int cpu) :
+gtpu_l4_stack::gtpu_l4_stack(const struct in_addr& address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params) :
     udp_s(udp_server(address, port_num))
 {
   Logger::gtpv1_u().info( "gtpu_l4_stack created listening to %s:%d", oai::cn::core::toString(address).c_str(), port_num);
@@ -205,10 +205,10 @@ gtpu_l4_stack::gtpu_l4_stack(const struct in_addr& address, const uint16_t port_
   srand (time(NULL));
   seq_num = rand() & 0x7FFFFFFF;
   restart_counter = 0;
-  udp_s.start_receive(this, cpu);
+  udp_s.start_receive(this, thread_sched_params);
 }
 //------------------------------------------------------------------------------
-gtpu_l4_stack::gtpu_l4_stack(const struct in6_addr& address, const uint16_t port_num, const int cpu) :
+gtpu_l4_stack::gtpu_l4_stack(const struct in6_addr& address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params) :
     udp_s(udp_server(address, port_num))
 {
   Logger::gtpv1_u().info( "gtpu_l4_stack created listening to %s:%d", oai::cn::core::toString(address).c_str(), port_num);
@@ -217,10 +217,10 @@ gtpu_l4_stack::gtpu_l4_stack(const struct in6_addr& address, const uint16_t port
   srand (time(NULL));
   seq_num = rand() & 0x7FFFFFFF;
   restart_counter = 0;
-  udp_s.start_receive(this, cpu);
+  udp_s.start_receive(this, thread_sched_params);
 }
 //------------------------------------------------------------------------------
-gtpu_l4_stack::gtpu_l4_stack(char * address, const uint16_t port_num, const int cpu) :
+gtpu_l4_stack::gtpu_l4_stack(char * address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params) :
         udp_s(udp_server(address, port_num))
 {
   Logger::gtpv1_u().info( "gtpu_l4_stack created listening to %s:%d", address, port_num);
@@ -229,7 +229,7 @@ gtpu_l4_stack::gtpu_l4_stack(char * address, const uint16_t port_num, const int
   srand (time(NULL));
   seq_num = rand() & 0x7FFFFFFF;
   restart_counter = 0;
-  udp_s.start_receive(this, cpu);
+  udp_s.start_receive(this, thread_sched_params);
 
 }
 

src/gtpv1u/gtpv1u.hpp

@@ -31,6 +31,7 @@
 #include "3gpp_29.281.hpp"
 #include "itti.hpp"
 #include "msg_gtpv1u.hpp"
+#include "thread_sched.h"
 
 #include <iostream>
 #include <map>
@@ -93,7 +94,7 @@ public:
     close(socket_);
   }
 
-  void udp_read_loop(int cpu_id);
+  void udp_read_loop(const thread_sched_params_t& thread_sched_params);
 
   void async_send_to(const char* send_buffer, const ssize_t num_bytes, const struct sockaddr_in& peer_addr)
   {
@@ -113,7 +114,7 @@ public:
   }
 
 
-  void start_receive(gtpu_l4_stack * gtp_stack, const int cpu);
+  void start_receive(gtpu_l4_stack * gtp_stack, const thread_sched_params_t& thread_sched_params);
 
 protected:
   int create_socket (const struct in_addr& address, const uint16_t port);
@@ -161,9 +162,9 @@ protected:
 
 public:
   static const uint8_t version = 1;
-  gtpu_l4_stack(const struct in_addr& address, const uint16_t port_num, const int cpu);
-  gtpu_l4_stack(const struct in6_addr& address, const uint16_t port_num, const int cpu);
-  gtpu_l4_stack(char * ip_address, const uint16_t port_num, const int cpu);
+  gtpu_l4_stack(const struct in_addr& address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params);
+  gtpu_l4_stack(const struct in6_addr& address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params);
+  gtpu_l4_stack(char * ip_address, const uint16_t port_num, const thread_sched_params_t& thread_sched_params);
   virtual void handle_receive(char* recv_buffer, const std::size_t bytes_transferred, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len);
   void handle_receive_message_cb(const gtpv1u_msg& msg, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len, const core::itti::task_id_t& task_id, bool &error, uint64_t& gtpc_tx_id);
 

src/spgwu/simpleswitch/pfcp_switch.hpp

+/*
+ * 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
+ */
+
+/*! \file pfcp_switch.hpp
+   \author  Lionel GAUTHIER
+   \date 2019
+   \email: lionel.gauthier@eurecom.fr
+*/
+
+#ifndef FILE_PFCP_SWITCH_HPP_SEEN
+#define FILE_PFCP_SWITCH_HPP_SEEN
+
+//#include "concurrentqueue.h"
+#include "itti.hpp"
+#include "itti_msg_sxab.hpp"
+#include "msg_pfcp.hpp"
+#include "pfcp_session.hpp"
+#include "uint_generator.hpp"
+#include "thread_sched.h"
+
+#include <folly/AtomicHashMap.h>
+#include <linux/ip.h>
+#include <linux/ipv6.h>
+#include <unordered_map>
+#include <memory>
+#include <netinet/in.h>
+#include <thread>
+#include <vector>
+
+namespace oai::cn::nf::spgwu {
+
+  // Have to be tuned for sdt situations
+#define PFCP_SWITCH_MAX_SESSIONS 512
+#define PFCP_SWITCH_MAX_PDRS     512
+#define PDN_INTERFACE_NAME      "pdn"
+//class switching_records{
+//public:
+//  switching_records() {
+//    cp_fseid2pfcp_sessions = {};
+//    up_seid2pfcp_sessions = {};
+//    ul_s1u_teid2pfcp_pdr = {};
+//    ue_ipv4_hbo2pfcp_pdr = {};
+//  }
+//  std::unordered_map<core::pfcp::fseid_t, std::shared_ptr<core::pfcp::pfcp_session>>                cp_fseid2pfcp_sessions;
+//  std::unordered_map<uint64_t, std::shared_ptr<core::pfcp::pfcp_session>>                           up_seid2pfcp_sessions;
+//};
+//
+//class switching_data_per_cpu_socket {
+//public:
+//  switching_data_per_cpu_socket() {
+//    msg_iov_= {};
+//    switching_up_traffic_index = 0;
+//    switching_control_index = 1;
+//    switching_records[0] = {};
+//    switching_records[1] = {};
+//  }
+//  switching_records                         switching_records[2];
+//  struct iovec                              msg_iov_;        /* scatter/gather array */
+//  uint8_t                                   switching_up_traffic_index;
+//  uint8_t                                   switching_control_index;
+//};
+
+class pfcp_switch {
+private:
+  // Very unoptimized
+#define PFCP_SWITCH_RECV_BUFFER_SIZE 2048
+#define ROOM_FOR_GTPV1U_G_PDU        64
+  char                                      recv_buffer_[PFCP_SWITCH_RECV_BUFFER_SIZE];
+  int                                       sock_r;
+  int                                       sock_w;
+  std::thread                               thread_sock_;
+  //std::string                               gw_mac_address;
+  int                                       pdn_if_index;
+
+
+  oai::cn::util::uint_generator<uint64_t>   seid_generator_;
+  oai::cn::util::uint_generator<teid_t>     teid_s1u_generator_;
+
+#define TASK_SPGWU_PFCP_SWITCH_MAX_COMMIT_INTERVAL     (0)
+#define TASK_SPGWU_PFCP_SWITCH_MIN_COMMIT_INTERVAL     (1)
+
+#define PFCP_SWITCH_MAX_COMMIT_INTERVAL_MILLISECONDS 200
+#define PFCP_SWITCH_MIN_COMMIT_INTERVAL_MILLISECONDS  50
+
+  //switching_data_per_cpu_socket             switching_data[];
+  struct iovec                              msg_iov_;        /* scatter/gather array */
+  std::unordered_map<core::pfcp::fseid_t, std::shared_ptr<core::pfcp::pfcp_session>>                  cp_fseid2pfcp_sessions;
+  folly::AtomicHashMap<uint64_t, std::shared_ptr<core::pfcp::pfcp_session>>                           up_seid2pfcp_sessions;
+  folly::AtomicHashMap<teid_t, std::shared_ptr<std::vector<std::shared_ptr<core::pfcp::pfcp_pdr>>>>   ul_s1u_teid2pfcp_pdr;
+  folly::AtomicHashMap<uint32_t, std::shared_ptr<std::vector<std::shared_ptr<core::pfcp::pfcp_pdr>>>> ue_ipv4_hbo2pfcp_pdr;
+
+  //moodycamel::ConcurrentQueue<core::pfcp::pfcp_session*>            create_session_q;
+
+  void pdn_read_loop(const thread_sched_params_t& thread_sched_params);
+  int create_pdn_socket (const char * const ifname, const bool promisc, int& if_index);
+  int create_pdn_socket (const char * const ifname);
+  void setup_pdn_interfaces();
+
+  oai::cn::core::itti::timer_id_t timer_max_commit_interval_id;
+  oai::cn::core::itti::timer_id_t timer_min_commit_interval_id;
+
+  void stop_timer_min_commit_interval();
+  void start_timer_min_commit_interval();
+  void stop_timer_max_commit_interval();
+  void start_timer_max_commit_interval();
+
+  void commit_changes();
+
+
+  bool get_pfcp_session_by_cp_fseid(const core::pfcp::fseid_t&, std::shared_ptr<core::pfcp::pfcp_session>&) const;
+  bool get_pfcp_session_by_up_seid(const uint64_t, std::shared_ptr<core::pfcp::pfcp_session>&) const;
+  bool get_pfcp_ul_pdrs_by_up_teid(const teid_t, std::shared_ptr<std::vector<std::shared_ptr<core::pfcp::pfcp_pdr>>>&) const;
+  bool get_pfcp_dl_pdrs_by_ue_ip(const uint32_t, std::shared_ptr<std::vector<std::shared_ptr<core::pfcp::pfcp_pdr>>>&) const;
+
+  void add_pfcp_session_by_cp_fseid(const core::pfcp::fseid_t&, std::shared_ptr<core::pfcp::pfcp_session>&);
+  void add_pfcp_session_by_up_seid(const uint64_t, std::shared_ptr<core::pfcp::pfcp_session>&);
+  void add_pfcp_ul_pdr_by_up_teid(const teid_t teid, std::shared_ptr<core::pfcp::pfcp_pdr>& );
+
+  void remove_pfcp_session(std::shared_ptr<core::pfcp::pfcp_session>&);
+
+
+
+  uint64_t generate_seid() {return seid_generator_.get_uid();};
+
+  teid_t generate_teid_s1u() {return teid_s1u_generator_.get_uid();};
+
+public:
+  pfcp_switch();
+  pfcp_switch(pfcp_switch const&)    = delete;
+  void operator=(pfcp_switch const&)     = delete;
+
+  void add_pfcp_dl_pdr_by_ue_ip(const uint32_t ue_ip, std::shared_ptr<core::pfcp::pfcp_pdr>& );
+
+  core::pfcp::fteid_t generate_fteid_s1u();
+  bool create_packet_in_access(std::shared_ptr<core::pfcp::pfcp_pdr>& pdr, const core::pfcp::fteid_t& in, uint8_t& cause);
+
+  void pfcp_session_look_up_pack_in_access(struct iphdr* const iph, const std::size_t num_bytes, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len, const uint32_t tunnel_id);
+  void pfcp_session_look_up_pack_in_access(struct ipv6hdr* const iph, const std::size_t num_bytes, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len, const uint32_t tunnel_id);
+  void pfcp_session_look_up_pack_in_access(struct iphdr* const iph, const std::size_t num_bytes, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len) {};
+  void pfcp_session_look_up_pack_in_access(struct ipv6hdr* const iph, const std::size_t num_bytes, const struct sockaddr_storage& r_endpoint, const socklen_t& r_endpoint_addr_len) {};
+  //void pfcp_session_look_up(struct ethhdr* const ethh, const std::size_t num_bytes);
+
+  void pfcp_session_look_up_pack_in_core(const char *buffer, const std::size_t num_bytes);
+
+  void send_to_core(char* const ip_packet, const ssize_t len);
+
+  void handle_pfcp_session_establishment_request(std::shared_ptr<core::itti::itti_sxab_session_establishment_request> sreq, core::itti::itti_sxab_session_establishment_response* );
+  void handle_pfcp_session_modification_request(std::shared_ptr<core::itti::itti_sxab_session_modification_request> sreq, core::itti::itti_sxab_session_modification_response* );
+  void handle_pfcp_session_deletion_request(std::shared_ptr<core::itti::itti_sxab_session_deletion_request> sreq, core::itti::itti_sxab_session_deletion_response* );
+
+  void time_out_min_commit_interval(const uint32_t timer_id);
+  void time_out_max_commit_interval(const uint32_t timer_id);
+
+  void remove_pfcp_session(const oai::cn::core::pfcp::fseid_t& cp_fseid);
+  void remove_pfcp_ul_pdrs_by_up_teid(const teid_t) {};
+  void remove_pfcp_dl_pdrs_by_ue_ip(const uint32_t) {};
+
+  std::string to_string() const;
+};
+}
+#endif /* FILE_PFCP_SWITCH_HPP_SEEN */

src/spgwu/simpleswitch/spgwu_s1u.cpp

@@ -55,6 +55,23 @@ void spgwu_s1u_task (void*);
 void spgwu_s1u_task (void *args_p)
 {
   const task_id_t task_id = TASK_SPGWU_S1U;
+  const thread_sched_params_t* const thread_sched_params = (const thread_sched_params_t* const)args_p;
+
+  if (thread_sched_params->cpu_id >= 0) {
+    cpu_set_t cpuset;
+    CPU_SET(thread_sched_params->cpu_id,&cpuset);
+    if (int rc = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset)) {
+      Logger::spgwu_s1u().warn( "Could not set affinity to ITTI task TASK_SPGWU_S1U, err=%d", rc);
+    }
+  }
+
+  struct sched_param sparam;
+  memset(&sparam, 0, sizeof(sparam));
+  sparam.sched_priority = thread_sched_params->sched_priority;
+  if (int rc = pthread_setschedparam(pthread_self(), thread_sched_params->sched_policy, &sparam)) {
+    Logger::spgwu_s1u().warn( "Could not set schedparam to ITTI task TASK_SPGWU_S1U, err=%d", rc);
+  }
+
   itti_inst->notify_task_ready(task_id);
 
   do {
@@ -88,10 +105,10 @@ void spgwu_s1u_task (void *args_p)
 }
 
 //------------------------------------------------------------------------------
-spgwu_s1u::spgwu_s1u () : gtpu_l4_stack(spgwu_cfg.s1_up.addr4, spgwu_cfg.s1_up.port, spgwu_cfg.s1_up.cpu_id_thread_loop_read)
+spgwu_s1u::spgwu_s1u () : gtpu_l4_stack(spgwu_cfg.s1_up.addr4, spgwu_cfg.s1_up.port, spgwu_cfg.s1_up.thread_rd_sched_params)
 {
   Logger::spgwu_s1u().startup("Starting...");
-  if (itti_inst->create_task(TASK_SPGWU_S1U, spgwu_s1u_task, nullptr) ) {
+  if (itti_inst->create_task(TASK_SPGWU_S1U, spgwu_s1u_task, &spgwu_cfg.itti.s1u_sched_params) ) {
     Logger::spgwu_s1u().error( "Cannot create task TASK_SPGWU_S1U" );
     throw std::runtime_error( "Cannot create task TASK_SPGWU_S1U" );
   }
@@ -238,7 +255,7 @@ void spgwu_s1u::report_error_indication(const struct sockaddr_storage& r_endpoin
     error_ind->gtp_ies.set(peer_address);
   } else {
     // mandatory ie
-    free(error_ind);
+    delete error_ind;
     return;
   }
 

src/spgwu/spgwu_config.cpp

@@ -95,6 +95,48 @@ int spgwu_config::get_pfcp_fseid(oai::cn::core::pfcp::fseid_t& fseid)
   }
   return rc;
 }
+//------------------------------------------------------------------------------
+int spgwu_config::load_thread_sched_params(const Setting& thread_sched_params_cfg, thread_sched_params_t& cfg)
+{
+
+  thread_sched_params_cfg.lookupValue(SPGWU_CONFIG_STRING_THREAD_RD_CPU_ID, cfg.cpu_id);
+  std::string thread_rd_sched_policy;
+  thread_sched_params_cfg.lookupValue(SPGWU_CONFIG_STRING_THREAD_RD_SCHED_POLICY, thread_rd_sched_policy);
+  util::trim(thread_rd_sched_policy);
+  if (boost::iequals(thread_rd_sched_policy, "SCHED_OTHER")) {
+    cfg.sched_policy = SCHED_OTHER;
+  } else if (boost::iequals(thread_rd_sched_policy, "SCHED_IDLE")) {
+    cfg.sched_policy = SCHED_IDLE;
+  } else if (boost::iequals(thread_rd_sched_policy, "SCHED_BATCH")) {
+    cfg.sched_policy = SCHED_BATCH;
+  } else if (boost::iequals(thread_rd_sched_policy, "SCHED_FIFO")) {
+    cfg.sched_policy = SCHED_FIFO;
+  } else if (boost::iequals(thread_rd_sched_policy, "SCHED_RR")) {
+    cfg.sched_policy = SCHED_RR;
+  } else {
+    Logger::spgwu_app().error("thread_rd_sched_policy: %s, unknown in config file", thread_rd_sched_policy.c_str());
+    return RETURNerror;
+  }
+  thread_sched_params_cfg.lookupValue(SPGWU_CONFIG_STRING_THREAD_RD_SCHED_PRIORITY, cfg.sched_priority);
+  if ((cfg.sched_priority > 99) || (cfg.sched_priority < 1)) {
+    Logger::spgwu_app().error("thread_rd_sched_priority: %d, must be in interval [1..99] in config file", cfg.sched_priority);
+    return RETURNerror;
+  }
+  return RETURNok;
+}
+//------------------------------------------------------------------------------
+int spgwu_config::load_itti(const Setting& itti_cfg, itti_cfg_t& cfg)
+{
+  const Setting& sched_params_cfg = itti_cfg[SPGWU_CONFIG_STRING_SCHED_PARAMS];
+  load_thread_sched_params(sched_params_cfg, cfg.core_sched_params);
+
+  const Setting& s1u_sched_params_cfg = itti_cfg[SPGWU_CONFIG_STRING_S1U_SCHED_PARAMS];
+  load_thread_sched_params(s1u_sched_params_cfg, cfg.s1u_sched_params);
+
+  const Setting& sx_sched_params_cfg = itti_cfg[SPGWU_CONFIG_STRING_SX_SCHED_PARAMS];
+  load_thread_sched_params(sx_sched_params_cfg, cfg.sx_sched_params);
+  return RETURNok;
+}
 
 //------------------------------------------------------------------------------
 int spgwu_config::load_interface(const Setting& if_cfg, interface_cfg_t& cfg)
@@ -127,7 +169,8 @@ int spgwu_config::load_interface(const Setting& if_cfg, interface_cfg_t& cfg)
       cfg.network4.s_addr = htons(ntohs(cfg.addr4.s_addr) & 0xFFFFFFFF << (32 - std::stoi (util::trim(words.at(1)))));
     }
     if_cfg.lookupValue(SPGWU_CONFIG_STRING_PORT, cfg.port);
-    if_cfg.lookupValue(SPGWU_CONFIG_STRING_CPU_ID_THREAD_LOOP_READ, cfg.cpu_id_thread_loop_read);
+    const Setting& sched_params_cfg = if_cfg[SPGWU_CONFIG_STRING_SCHED_PARAMS];
+    load_thread_sched_params(sched_params_cfg, cfg.thread_rd_sched_params);
   }
   return RETURNok;
 }
@@ -165,6 +208,9 @@ int spgwu_config::load(const string& config_file)
     spgwu_cfg.lookupValue(SPGWU_CONFIG_STRING_PID_DIRECTORY, pid_dir);
     util::trim(pid_dir);
 
+    const Setting& itti_cfg = spgwu_cfg[SPGWU_CONFIG_STRING_ITTI];
+    load_itti(itti_cfg, itti);
+
     const Setting& nw_if_cfg = spgwu_cfg[SPGWU_CONFIG_STRING_INTERFACES];
 
     const Setting& s1_up_cfg = nw_if_cfg[SPGWU_CONFIG_STRING_INTERFACE_S1U_S12_S4_UP];

src/spgwu/spgwu_config.hpp

@@ -30,6 +30,7 @@
 #define FILE_SPGWU_CONFIG_HPP_SEEN
 
 #include "3gpp_29.244.h"
+#include "thread_sched.h"
 #include <libconfig.h++>
 #include <mutex>
 #include <netinet/in.h>
@@ -47,7 +48,12 @@ namespace oai::cn::nf::spgwu {
 #define SPGWU_CONFIG_STRING_INTERFACE_NAME                        "INTERFACE_NAME"
 #define SPGWU_CONFIG_STRING_IPV4_ADDRESS                          "IPV4_ADDRESS"
 #define SPGWU_CONFIG_STRING_PORT                                  "PORT"
-#define SPGWU_CONFIG_STRING_CPU_ID_THREAD_LOOP_READ               "CPU_ID_THREAD_LOOP_READ"
+#define SPGWU_CONFIG_STRING_SCHED_PARAMS                          "SCHED_PARAMS"
+#define SPGWU_CONFIG_STRING_S1U_SCHED_PARAMS                      "S1U_SCHED_PARAMS"
+#define SPGWU_CONFIG_STRING_SX_SCHED_PARAMS                       "SX_SCHED_PARAMS"
+#define SPGWU_CONFIG_STRING_THREAD_RD_CPU_ID                      "THREAD_RD_CPU_ID"
+#define SPGWU_CONFIG_STRING_THREAD_RD_SCHED_POLICY                "THREAD_RD_SCHED_POLICY"
+#define SPGWU_CONFIG_STRING_THREAD_RD_SCHED_PRIORITY              "THREAD_RD_SCHED_PRIORITY"
 #define SPGWU_CONFIG_STRING_INTERFACE_SGI                         "SGI"
 #define SPGWU_CONFIG_STRING_INTERFACE_SX                          "SX"
 #define SPGWU_CONFIG_STRING_INTERFACE_S1U_S12_S4_UP               "S1U_S12_S4_UP"
@@ -58,6 +64,7 @@ namespace oai::cn::nf::spgwu {
 #define SPGWU_CONFIG_STRING_SNAT                                  "SNAT"
 #define SPGWU_CONFIG_STRING_MAX_PFCP_SESSIONS                     "MAX_PFCP_SESSIONS"
 #define SPGWU_CONFIG_STRING_SPGWC_LIST                            "SPGW-C_LIST"
+#define SPGWU_CONFIG_STRING_ITTI                                  "ITTI"
 
 #define SPGW_ABORT_ON_ERROR true
 #define SPGW_WARN_ON_ERROR false
@@ -69,7 +76,7 @@ typedef struct interface_cfg_s {
   struct in6_addr addr6;
   unsigned int    mtu;
   unsigned int    port;
-  int             cpu_id_thread_loop_read;
+  thread_sched_params_t thread_rd_sched_params;
 } interface_cfg_t;
 
 typedef struct pdn_cfg_s {
@@ -80,11 +87,19 @@ typedef struct pdn_cfg_s {
   bool            snat;
 } pdn_cfg_t;
 
+typedef struct itti_cfg_s {
+  thread_sched_params_t core_sched_params;
+  thread_sched_params_t s1u_sched_params;
+  thread_sched_params_t sx_sched_params;
+} itti_cfg_t;
+
 class spgwu_config {
 
 private:
 
+  int load_itti(const libconfig::Setting& itti_cfg, itti_cfg_t& cfg);
   int load_interface(const libconfig::Setting& if_cfg, interface_cfg_t& cfg);
+  int load_thread_sched_params(const libconfig::Setting& thread_sched_params_cfg, thread_sched_params_t& cfg);
 
 public:
 
@@ -95,6 +110,8 @@ public:
   interface_cfg_t s1_up;
   interface_cfg_t sgi;
   interface_cfg_t sx;
+  itti_cfg_t      itti;
+
   std::string gateway;
 
   uint32_t        max_pfcp_sessions;
@@ -103,7 +120,7 @@ public:
   std::vector<oai::cn::core::pfcp::node_id_t> spgwcs;
 
 
-  spgwu_config() : m_rw_lock(), pid_dir(), instance(0), s1_up(), sgi(), gateway(),sx(), pdns(), spgwcs(), max_pfcp_sessions(100) {};
+  spgwu_config() : m_rw_lock(), pid_dir(), instance(0), s1_up(), sgi(), gateway(), sx(), itti(), pdns(), spgwcs(), max_pfcp_sessions(100) {};
   void lock() {m_rw_lock.lock();};
   void unlock() {m_rw_lock.unlock();};
   int load(const std::string& config_file);