Single executable includes coordinator (#207)

* Single binary GNB-CU with coordinator

* Progress towards single executable model

* Tidying

README.md

@@ -6,6 +6,17 @@ The gNodeB is the component that manages the radio access of 5G User Equipment (
 
 This project is currently a proof of concept and not yet a fully functional gNB-CU.
 
+## Quickstart
+```
+cargo build
+./target/debug/gnb-cu --mcc 111 --mnc 11 --amf-ip 5.5.5.5 &
+./target/debug/gnb-cu --mcc 111 --mnc 11 --amf-ip 5.5.5.5
+```
+
+This uses the simplified startup model for running on a single host.  The first instance
+binds to 127.0.0.1.  The second binds to 127.0.0.2.
+
+
 ## Current support
 - UE registration demo against free5GC.
 - PDU session setup and release (TS 23.502, figures 4.3.2.2.1-1 and 4.3.4.2-1).

common/Cargo.toml

@@ -17,7 +17,7 @@ anyhow = "1.0.52"
 async-channel = "1.6.1"
 libc = "^0.2"
 async-io = "1.7.0"
-futures-lite = "1.12.0"
+futures-lite = "1.13.0"
 log = "0.4.3"
 futures-core = "0.3.19"
 slog-envlogger = "2.2.0"

documentation/backlog.md

@@ -1,10 +1,20 @@
-# Single process GNB-CU
-
 # NEXT UP
+- useless error : 
+Jul 19 06:38:12.966 WARN Failing all requests because of TNLA 17 termination. Note that current blanket implementation may drop requests on other TNLAs that could have survived, cu-cp: 1
+Jul 19 06:38:12.966 INFO NGAP TNLA 17 closed, cu-cp: 1
+Jul 19 06:38:12.966 WARN Channel recv error: RecvError, cu-cp: 1
+- Ctrl-C is not reliably terminating
+- in RequestProvider, pass a transaction context instead of a logger
+  - transaciton context provides logger
+  - transaciotn context also provides "response action" to simplify?
+  - also provides TNLA ID and remote IP address - can use this to remove double info! logging of TNLA setup
+- userplane HA (two E1AP connections and userplane replication)
+- warn! consistently on error in workflow
 - RRC connection release
 - UE context release
 - Testing of Session/context releases on different worker
 - Paging
+- promotion of coordinator in single executable mode
 
 # TECH DEBT
 ## CU-UP and O-RAN O-DU interop
@@ -41,7 +51,6 @@
 ## FUNCTION
 - Proper graceful shutdown (waiting for / sending responses to pending requests)
 - Make values in NG Setup configurable rather than hard coded (Tac, Plmn Id, slices, etc)
-- Generate RRC transaction IDs properly
 - Don't hang indefinitely waiting for response (e.g. NG Setup response)
 - Don't allow unlimited pending requests
 - Handle -ve response to InitialContextSetupRequest with bad RAN UE ID
@@ -90,6 +99,7 @@
 - Distributed timers and failure path cleanup mechanism
 
 # DONE
+- Add coordinator to single process GNB-CU
 - PDU session deletion
 - Retry connection to AMF
 - Single process GNB-CU with configurable MCC / MNC

documentation/design/Coordinator autoconfiguration.md

@@ -0,0 +1,5 @@
+
+Problems
+- choose a local address (127.0.0.1, 127.0.0.2)
+- race to bind a coordinator to 0.0.0.0
+  - race to run a local coordinator in case no coordinator is found

e1ap/src/top_pdu.rs

@@ -5,7 +5,7 @@ use crate::common::Criticality;
 use anyhow::Result;
 use asn1_per::{aper::*, *};
 use async_trait::async_trait;
-use slog::Logger;
+use slog::{error, Logger};
 
 // E1apPdu
 #[derive(Clone, Debug)]
@@ -278,7 +278,13 @@ impl Procedure for GnbCuCpConfigurationUpdateProcedure {
                 ),
                 f,
             )),
-            Err(_) => todo!(),
+            Err(_) => {
+                error!(
+                    logger,
+                    "Error path not implemented for GnbCuCpConfigurationUpdateProcedure"
+                );
+                None
+            }
         }
     }
 

gnb-cu-cp-coordinator/src/server.rs

@@ -91,7 +91,7 @@ impl<C> Server<C> {
 }
 
 pub fn spawn(config: Config, logger: Logger) -> Result<ShutdownHandle> {
-    info!(logger, "Started");
+    debug!(logger, "Started");
     let stop_source = StopSource::new();
     let stop_token = stop_source.token();
 

gnb-cu-cp/src/main.rs

@@ -40,7 +40,8 @@ async fn main() -> Result<()> {
         config,
         RedisUeStore::new(6379).unwrap(),
         root_logger.clone(),
-    )?;
+    )
+    .await?;
     let s = signal::wait_for_signal().await?;
     info!(root_logger, "Caught signal {} - terminate", s);
     shutdown_handle.graceful_shutdown().await;

gnb-cu-cp/src/worker.rs

@@ -70,7 +70,7 @@ const F1AP_BIND_PORT: u16 = 38472;
 const E1AP_SCTP_PPID: u32 = 64;
 const E1AP_BIND_PORT: u16 = 38462;
 
-pub fn spawn<U: UeStateStore>(
+pub async fn spawn<U: UeStateStore>(
     worker_id: Uuid,
     config: Config,
     ue_store: U,
@@ -81,6 +81,7 @@ pub fn spawn<U: UeStateStore>(
 
     info!(&logger, "Starting gNB-CU-CP worker {}", worker_id);
     info!(&logger, "PLMN is {:02x?}", config.plmn);
+    debug!(&logger, "Config: {:?}", config);
 
     let handle = match config.connection_style {
         // Run a combined worker and coordinator.
@@ -115,11 +116,9 @@ pub fn spawn<U: UeStateStore>(
             )
             .unwrap();
             let worker = Worker::new(config, ue_store, worker_id, logger, coordinator);
+            worker.start_servers().await?;
             async_std::task::spawn(async move {
-                worker
-                    .serve(stop_token)
-                    .await
-                    .expect("Worker startup failure");
+                worker.run(stop_token).await;
             })
         }
     };
@@ -150,7 +149,7 @@ impl<A: Clone + Send + Sync + 'static + CoordinationApi<ClientContext>, U: UeSta
         }
     }
 
-    async fn serve(self, stop_token: StopToken) -> Result<()> {
+    async fn start_servers(&self) -> Result<()> {
         let f1ap_handle = self.serve_f1ap().await?;
         self.add_shutdown_handle(f1ap_handle).await;
 
@@ -165,7 +164,10 @@ impl<A: Clone + Send + Sync + 'static + CoordinationApi<ClientContext>, U: UeSta
             let connection_api_handle = self.serve_connection_api(connection_api_bind_port).await?;
             self.add_shutdown_handle(connection_api_handle).await;
         };
+        Ok(())
+    }
 
+    async fn run(self, stop_token: StopToken) {
         // Connect to the coordinator.  It will bring this worker into service by making calls to the
         // connection API.
         self.send_periodic_refreshes_to_coordinator(stop_token.clone())
@@ -178,7 +180,11 @@ impl<A: Clone + Send + Sync + 'static + CoordinationApi<ClientContext>, U: UeSta
         }
 
         self.ngap.graceful_shutdown().await;
+    }
 
+    async fn serve(self, stop_token: StopToken) -> Result<()> {
+        self.start_servers().await?;
+        self.run(stop_token).await;
         Ok(())
     }
 
@@ -283,7 +289,7 @@ impl<A: Clone + Send + Sync + 'static + CoordinationApi<ClientContext>, U: UeSta
         let connection_api_listen_address = self.worker_listen_address(port);
         info!(
             &self.logger,
-            "Serve connection API on {}", connection_api_listen_address
+            "Serve connection API on {connection_api_listen_address}",
         );
         let addr = connection_api_listen_address.parse()?;
         crate::handlers::connection_api::serve(addr, self.clone(), self.logger.clone()).await

gnb-cu-up/Cargo.toml

@@ -21,3 +21,4 @@ asn1-per = { path = "../asn1-per" }
 async-net = "1.6.1"
 dashmap = "5.4.0"
 rand = "0.8.5"
+socket2 = { version = "0.5.3", features = ["all"] }

gnb-cu-up/src/main.rs

@@ -9,7 +9,7 @@ async fn main() -> Result<()> {
 
     let root_logger = logging::init();
     let config = Config::default();
-    let shutdown_handle = gnb_cu_up::spawn(config, root_logger.clone())?;
+    let shutdown_handle = gnb_cu_up::spawn(config, root_logger.clone()).await?;
     let s = signal::wait_for_signal().await?;
     info!(root_logger, "Caught signal {} - terminate", s);
     shutdown_handle.graceful_shutdown().await;

gnb-cu-up/src/packet_processor.rs

@@ -1,13 +1,15 @@
 #![allow(clippy::unusual_byte_groupings)]
 use std::sync::Arc;
 
-use anyhow::Result;
-use async_net::{IpAddr, SocketAddr, UdpSocket};
+use anyhow::{ensure, Context, Result};
+use async_net::{IpAddr, UdpSocket};
 use async_std::{
     sync::Mutex,
     task::{self, JoinHandle},
 };
-use slog::{debug, Logger};
+use slog::{debug, info, Logger};
+use socket2::{Domain, Protocol, Socket, Type};
+use std::net::SocketAddr;
 use xxap::{GtpTeid, GtpTunnel};
 
 const GTPU_PORT: u16 = 2152; // TS29.281
@@ -43,8 +45,19 @@ const CAPACITY_MASK: u32 = 0xff;
 impl PacketProcessor {
     pub async fn new(local_ip: IpAddr, logger: Logger) -> Result<Self> {
         let transport_address = SocketAddr::new(local_ip, GTPU_PORT);
-        let gtpu_socket = UdpSocket::bind(transport_address).await?;
-        // See set_reuse_address in socket2 create.  May be necessary for HA.
+        let domain = match local_ip {
+            IpAddr::V4(_) => Domain::IPV4,
+            IpAddr::V6(_) => Domain::IPV6,
+        };
+        ensure!(matches!(local_ip, IpAddr::V4(_)));
+        let gtpu_socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
+        gtpu_socket.set_reuse_port(true)?;
+        info!(logger, "Serving GTP-U on {transport_address}");
+        gtpu_socket
+            .bind(&transport_address.into())
+            .context(format!("Failed to bind {}", transport_address))?;
+        let gtpu_socket: std::net::UdpSocket = gtpu_socket.into();
+        let gtpu_socket = UdpSocket::try_from(gtpu_socket)?;
 
         let forwarding_table = Arc::new(Mutex::new(ForwardingTable(vec![
             ForwardingContext {

gnb-cu-up/src/worker.rs

@@ -36,15 +36,13 @@ pub struct Worker {
 const E1AP_SCTP_PPID: u32 = 64;
 const E1AP_BIND_PORT: u16 = 38462;
 
-pub fn spawn(config: Config, logger: Logger) -> Result<ShutdownHandle> {
+pub async fn spawn(config: Config, logger: Logger) -> Result<ShutdownHandle> {
     let stop_source = StopSource::new();
     let stop_token = stop_source.token();
     info!(&logger, "Starting gNB-CU-UP worker");
-    let handle = async_std::task::spawn(async move {
-        let worker = Worker::new(config, logger.clone())
-            .await
-            .expect("Worker startup failure");
+    let worker = Worker::new(config, logger.clone()).await?;
 
+    let handle = async_std::task::spawn(async move {
         worker
             .run(stop_token)
             .await

gnb-cu/Cargo.toml

@@ -15,3 +15,4 @@ gnb-cu-cp = {path = "../gnb-cu-cp" }
 gnb-cu-up = {path = "../gnb-cu-up" }
 uuid = {version = "1.3", features = ["serde", "v4"]}
 clap = { version = "4.1.6", features = ["derive"] }
+coordinator = {path = "../gnb-cu-cp-coordinator", package = "gnb-cu-cp-coordinator" }