fixeria
/
srsRAN
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

cleaned up socket util methods, and made the S1AP non-blocking.

This commit is contained in:
Francisco Paisana 2019-11-15 11:30:42 +00:00
parent 2c3e4a3daf
commit 76673d2d40
11 changed files with 1048 additions and 858 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

187
lib/include/srslte/common/network_utils.h
View File

@ -36,144 +36,137 @@
namespace srslte {
class rx_socket_itf_t
{
virtual int read(void* buf, size_t nbytes) const = 0;
};
namespace net_utils {
class tx_socket_itf_t
{
virtual int send(const void* buf, size_t nbytes) const = 0;
};
enum class addr_family { ipv4 = AF_INET, ipv6 = AF_INET6 };
enum class socket_type : int { none = -1, datagram = SOCK_DGRAM, stream = SOCK_STREAM, seqpacket = SOCK_SEQPACKET };
enum class protocol_type : int { NONE = -1, SCTP = IPPROTO_SCTP, TCP = IPPROTO_TCP, UDP = IPPROTO_UDP };
enum class ppid_values : uint32_t { S1AP = 18 };
const char* protocol_to_string(protocol_type p);
class net_addr_t
{
public:
std::string ip() const;
bool set_ip(const char* ip_str);
void set_port(int port) { addr.sin_port = port; }
int port() const { return addr.sin_port; }
const sockaddr_in& get_sockaddr_in() const { return addr; }
sockaddr_in& get_sockaddr_in() { return addr; }
// Convenience methods
bool set_sockaddr(sockaddr_in* addr, const char* ip_str, int port);
std::string get_ip(const sockaddr_in& addr);
int get_port(const sockaddr_in& addr);
net_utils::socket_type get_addr_family(int fd);
private:
struct sockaddr_in addr = {};
};
} // namespace net_utils
/**
* Description: Class created for code reuse by different sockets
* Description: Net socket class with convenience methods for connecting, binding, and opening socket
*/
class base_socket_t
class socket_handler_t
{
public:
base_socket_t() = default;
base_socket_t(const base_socket_t&) = delete;
base_socket_t(base_socket_t&& other) noexcept;
virtual ~base_socket_t();
base_socket_t& operator=(const base_socket_t&) = delete;
base_socket_t& operator =(base_socket_t&&) noexcept;
socket_handler_t() = default;
socket_handler_t(const socket_handler_t&) = delete;
socket_handler_t(socket_handler_t&& other) noexcept;
~socket_handler_t();
socket_handler_t& operator=(const socket_handler_t&) = delete;
socket_handler_t& operator =(socket_handler_t&&) noexcept;
bool is_init() const { return sockfd >= 0; }
int fd() const { return sockfd; }
void close();
void reset();
bool is_init() const { return sockfd >= 0; }
int fd() const { return sockfd; }
const sockaddr_in& get_addr_in() const { return addr; }
std::string get_ip() const { return net_utils::get_ip(addr); }
net_utils::socket_type get_family() const { return net_utils::get_addr_family(sockfd); }
bool bind_addr(const char* bind_addr_str, int port, srslte::log* log_ = nullptr);
bool connect_to(const char* dest_addr_str,
int dest_port,
sockaddr_in* dest_sockaddr = nullptr,
srslte::log* log_ = nullptr);
bool open_socket(net_utils::addr_family ip,
net_utils::socket_type socket_type,
net_utils::protocol_type protocol,
srslte::log* log_ = nullptr);
protected:
void reset_();
int bind_addr(const char* bind_addr_str, int port);
virtual int create_socket() = 0;
int connect_to(struct sockaddr_in* dest_addr, const char* dest_addr_str, int dest_port);
int sockfd = -1;
struct sockaddr_in addr_in = {};
sockaddr_in addr = {};
int sockfd = -1;
};
/**
* Description: handles the lifetime of a SCTP socket and provides convenience methods for listening/connecting, and
* read/send
*/
class sctp_socket_t final : public base_socket_t
{
public:
void reset();
int listen_addr(const char* bind_addr_str, int port);
int connect_addr(const char* bind_addr_str, const char* dest_addr_str, int dest_port);
namespace net_utils {
int read(void* buf, size_t nbytes, net_addr_t* addr) const;
int read(void* buf,
size_t nbytes,
struct sockaddr_in* from = nullptr,
socklen_t* fromlen = nullptr,
struct sctp_sndrcvinfo* sinfo = nullptr,
int msg_flags = 0) const;
int send(void* buf, size_t nbytes, uint32_t ppid, uint32_t stream_id) const;
bool sctp_init_client(socket_handler_t* socket,
net_utils::socket_type socktype,
const char* bind_addr_str,
srslte::log* log_);
bool sctp_init_server(socket_handler_t* socket,
net_utils::socket_type socktype,
const char* bind_addr_str,
int port,
srslte::log* log_);
private:
int create_socket() override;
// TODO: for TCP and UDP
bool tcp_make_server(socket_handler_t* socket,
const char* bind_addr_str,
int port,
int nof_connections = 1,
srslte::log* log_ = nullptr);
int tcp_accept(socket_handler_t* socket, sockaddr_in* destaddr, srslte::log* log_);
int tcp_read(int remotefd, void* buf, size_t nbytes, srslte::log* log_);
int tcp_send(int remotefd, const void* buf, size_t nbytes, srslte::log* log_);
struct sockaddr_in dest_addr = {};
};
} // namespace net_utils
class tcp_socket_t final : public base_socket_t, public rx_socket_itf_t, public tx_socket_itf_t
{
public:
void reset();
int listen_addr(const char* bind_addr_str, int port);
int accept_connection();
int connect_addr(const char* bind_addr_str, const char* dest_addr_str, int dest_port);
int read(void* buf, size_t nbytes) const override;
int send(const void* buf, size_t nbytes) const override;
private:
int create_socket() override;
struct sockaddr_in dest_addr = {};
int connfd = -1;
};
/****************************
* Rx multisocket handler
***************************/
class rx_multisocket_handler final : public thread
{
public:
using sctp_callback_t = std::function<void(const sctp_socket_t&)>;
using tcp_callback_t = std::function<void(const tcp_socket_t&)>;
// polymorphic callback to handle the socket recv
class recv_task
{
public:
virtual ~recv_task() = default;
virtual bool operator()(int fd) = 0; // returns false, if socket needs to be removed
};
using task_callback_t = std::unique_ptr<recv_task>;
using recv_callback_t = std::function<void(srslte::unique_byte_buffer_t)>;
using sctp_recv_callback_t =
std::function<void(srslte::unique_byte_buffer_t, const sockaddr_in&, const sctp_sndrcvinfo&, int)>;
rx_multisocket_handler(std::string name_, srslte::log* log_);
rx_multisocket_handler(std::string name_, srslte::log* log_, int thread_prio = 65);
rx_multisocket_handler(rx_multisocket_handler&&) = delete;
rx_multisocket_handler(const rx_multisocket_handler&) = delete;
rx_multisocket_handler& operator=(const rx_multisocket_handler&) = delete;
rx_multisocket_handler& operator=(const rx_multisocket_handler&&) = delete;
~rx_multisocket_handler();
template <typename Sock, typename Handler>
bool register_socket(const Sock& s, Handler&& handler)
{
auto func = [&s, handler]() { handler(s); };
return register_socket_(std::pair<const int, std::function<void()> >(s.fd(), func));
}
// bool register_sctp_socket(const sctp_socket_t& sock, const sctp_callback_t& recv_handler_);
// bool register_tcp_socket(const tcp_socket_t& sock, const tcp_callback_t& recv_handler_);
void stop();
bool remove_socket(int fd);
bool add_socket_handler(int fd, task_callback_t handler);
// convenience methods for recv using buffer pool
bool add_socket_pdu_handler(int fd, recv_callback_t pdu_task);
bool add_socket_sctp_handler(int fd, sctp_recv_callback_t task);
void run_thread() override;
private:
const static int THREAD_PRIO = 65;
// used to unlock select
struct ctrl_cmd_t {
enum class cmd_id_t { EXIT, NEW_FD };
enum class cmd_id_t { EXIT, NEW_FD, RM_FD };
cmd_id_t cmd = cmd_id_t::EXIT;
int new_fd = -1;
};
bool register_socket_(std::pair<const int, std::function<void()> >&& elem);
bool remove_socket_unprotected(int fd, fd_set* total_fd_set, int* max_fd);
// args
std::string name;
srslte::log* log_h = nullptr;
std::string name;
srslte::log* log_h = nullptr;
srslte::byte_buffer_pool* pool = nullptr;
// state
std::mutex socket_mutex;
std::map<int, std::function<void()> > active_sockets;
bool running = false;
int pipefd[2] = {};
std::mutex socket_mutex;
std::map<int, task_callback_t> active_sockets;
bool running = false;
int pipefd[2] = {};
};
} // namespace srslte

2
lib/include/srslte/common/timers.h
View File

@ -128,10 +128,10 @@ class timer_handler
void trigger()
{
if (is_running()) {
running = false;
if (callback) {
callback(id());
}
running = false;
}
}
};

10
lib/include/srslte/interfaces/enb_interfaces.h
View File

@ -340,6 +340,14 @@ typedef struct {
sched_interface::sched_args_t sched;
int link_failure_nof_err;
} mac_args_t;
}
class stack_interface_s1ap_lte
{
public:
virtual void add_mme_socket(int fd) = 0;
virtual void remove_mme_socket(int fd) = 0;
};
} // namespace srsenb
#endif // SRSLTE_ENB_INTERFACES_H

590
lib/src/common/network_utils.cc
View File

@ -22,331 +22,478 @@
#include "srslte/common/network_utils.h"
#include <netinet/sctp.h>
#include <sys/socket.h>
#include <sys/types.h>
#define rxSockError(fmt, ...) log_h->error("%s: " fmt, name.c_str(), ##__VA_ARGS__)
#define rxSockWarn(fmt, ...) log_h->warning("%s: " fmt, name.c_str(), ##__VA_ARGS__)
#define rxSockInfo(fmt, ...) log_h->info("%s: " fmt, name.c_str(), ##__VA_ARGS__)
#define rxSockDebug(fmt, ...) log_h->debug("%s: " fmt, name.c_str(), ##__VA_ARGS__)
namespace srslte {
std::string net_addr_t::ip() const
namespace net_utils {
bool set_sockaddr(sockaddr_in* addr, const char* ip_str, int port)
{
char ip_str[128];
// TODO: check whether IP4 or IP6 based on provided input
addr->sin_family = AF_INET;
if (inet_pton(AF_INET, ip_str, &addr->sin_addr) != 1) {
perror("inet_pton");
return false;
}
addr->sin_port = (port != 0) ? htons(port) : 0;
return true;
}
std::string get_ip(const sockaddr_in& addr)
{
char ip_str[128]; // TODO: check max size
inet_ntop(addr.sin_family, &addr.sin_addr, ip_str, sizeof(ip_str));
return std::string{ip_str};
}
bool net_addr_t::set_ip(const char* ip_str)
int get_port(const sockaddr_in& addr)
{
addr.sin_family = AF_INET; // ip4 only for now
if (inet_pton(AF_INET, ip_str, &addr.sin_addr) != 1) {
perror("inet_pton");
return false;
}
return true;
return ntohs(addr.sin_port);
}
net_utils::socket_type get_addr_family(int fd)
{
if (fd < 0) {
return net_utils::socket_type::none;
}
int type;
socklen_t length = sizeof(int);
getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &length);
return (net_utils::socket_type)type;
}
const char* protocol_to_string(protocol_type p)
{
switch (p) {
case protocol_type::TCP:
return "TCP";
case protocol_type::UDP:
return "UDP";
case protocol_type::SCTP:
return "SCTP";
default:
break;
}
return "";
}
} // namespace net_utils
/********************************************
* Socket Classes
*******************************************/
base_socket_t::base_socket_t(base_socket_t&& other) noexcept
socket_handler_t::socket_handler_t(socket_handler_t&& other) noexcept
{
sockfd = other.sockfd;
memcpy(&addr_in, &other.addr_in, sizeof(addr_in));
sockfd = other.sockfd;
addr = other.addr;
other.sockfd = 0;
bzero(&other.addr_in, sizeof(other.addr_in));
other.addr = {};
}
base_socket_t::~base_socket_t()
socket_handler_t::~socket_handler_t()
{
if (sockfd >= 0) {
close(sockfd);
}
reset();
}
base_socket_t& base_socket_t::operator=(base_socket_t&& other) noexcept
socket_handler_t& socket_handler_t::operator=(socket_handler_t&& other) noexcept
{
if (this == &other) {
return *this;
}
memcpy(&addr_in, &other.addr_in, sizeof(addr_in));
sockfd = other.sockfd;
bzero(&other.addr_in, sizeof(other.addr_in));
addr = other.addr;
sockfd = other.sockfd;
other.addr = {};
other.sockfd = 0;
return *this;
}
void base_socket_t::reset_()
void socket_handler_t::close()
{
if (sockfd >= 0) {
close(sockfd);
::close(sockfd);
sockfd = -1;
}
addr_in = {};
}
int base_socket_t::bind_addr(const char* bind_addr_str, int port)
void socket_handler_t::reset()
{
this->close();
addr = {};
}
bool socket_handler_t::bind_addr(const char* bind_addr_str, int port, srslte::log* log_)
{
if (sockfd < 0) {
if (create_socket() != 0) {
return -1;
if (log_ != nullptr) {
log_->error("Trying to bind to a closed socket\n");
}
return false;
}
addr_in.sin_family = AF_INET;
addr_in.sin_port = (port != 0) ? htons(port) : 0;
if (inet_pton(AF_INET, bind_addr_str, &(addr_in.sin_addr)) != 1) {
perror("inet_pton");
return -1;
if (not net_utils::set_sockaddr(&addr, bind_addr_str, port)) {
if (log_ != nullptr) {
log_->error("Failed to convert IP address (%s) to sockaddr_in struct\n", bind_addr_str);
}
return false;
}
if (bind(sockfd, (struct sockaddr*)&addr_in, sizeof(addr_in)) != 0) {
perror("bind()");
return -1;
if (bind(sockfd, (struct sockaddr*)&addr, sizeof(addr)) != 0) {
if (log_ != nullptr) {
log_->error("Failed to bind on address %s: %s errno %d\n", bind_addr_str, strerror(errno), errno);
}
return false;
}
return 0;
return true;
}
int base_socket_t::connect_to(struct sockaddr_in* dest_addr, const char* dest_addr_str, int dest_port)
bool socket_handler_t::connect_to(const char* dest_addr_str,
int dest_port,
sockaddr_in* dest_sockaddr,
srslte::log* log_)
{
dest_addr->sin_family = AF_INET;
dest_addr->sin_port = htons(dest_port);
if (inet_pton(AF_INET, dest_addr_str, &(dest_addr->sin_addr)) != 1) {
perror("inet_pton()");
return -1;
if (sockfd < 0) {
if (log_ != nullptr) {
log_->error("tried to connect to remote address with a closed socket.\n");
}
return false;
}
if (connect(sockfd, (struct sockaddr*)dest_addr, sizeof(*dest_addr)) == -1) {
perror("connect()");
return -1;
sockaddr_in sockaddr_tmp{};
sockaddr_in* sockaddr_ptr = (dest_sockaddr == nullptr) ? &sockaddr_tmp : dest_sockaddr;
*sockaddr_ptr = {};
if (not net_utils::set_sockaddr(sockaddr_ptr, dest_addr_str, dest_port)) {
if (log_ != nullptr) {
log_->error("Error converting IP address (%s) to sockaddr_in structure\n", dest_addr_str);
}
return false;
}
return 0;
if (connect(sockfd, (const struct sockaddr*)sockaddr_ptr, sizeof(*sockaddr_ptr)) == -1) {
if (log_ != nullptr) {
log_->error("Failed to establish socket connection to %s\n", dest_addr_str);
}
return false;
}
return true;
}
bool socket_handler_t::open_socket(net_utils::addr_family ip_type,
net_utils::socket_type socket_type,
net_utils::protocol_type protocol,
srslte::log* log_)
{
if (sockfd >= 0) {
if (log_ != nullptr) {
log_->error("Socket is already open.\n");
}
return false;
}
sockfd = socket((int)ip_type, (int)socket_type, (int)protocol);
if (sockfd == -1) {
if (log_ != nullptr) {
log_->error("Failed to open %s socket.\n", net_utils::protocol_to_string(protocol));
}
perror("Could not create socket\n");
return false;
}
return true;
}
/***********************************************************************
* SCTP socket
**********************************************************************/
void sctp_socket_t::reset()
{
reset_();
dest_addr = {};
}
namespace net_utils {
int sctp_socket_t::listen_addr(const char* bind_addr_str, int port)
bool sctp_init_socket(socket_handler_t* socket,
net_utils::socket_type socktype,
const char* bind_addr_str,
int port,
srslte::log* log_)
{
if (sockfd < 0 and create_socket() != 0) {
reset();
return SRSLTE_ERROR;
if (not socket->open_socket(net_utils::addr_family::ipv4, socktype, net_utils::protocol_type::SCTP, log_)) {
return false;
}
// Sets the data_io_event to be able to use sendrecv_info
// Subscribes to the SCTP_SHUTDOWN event, to handle graceful shutdown
struct sctp_event_subscribe evnts = {};
evnts.sctp_data_io_event = 1;
evnts.sctp_shutdown_event = 1;
if (setsockopt(sockfd, IPPROTO_SCTP, SCTP_EVENTS, &evnts, sizeof(evnts)) != 0) {
if (setsockopt(socket->fd(), IPPROTO_SCTP, SCTP_EVENTS, &evnts, sizeof(evnts)) != 0) {
perror("setsockopt");
reset();
return SRSLTE_ERROR;
socket->reset();
return false;
}
// bind addr
if (bind_addr(bind_addr_str, port) != 0) {
reset();
return SRSLTE_ERROR;
if (not socket->bind_addr(bind_addr_str, port, log_)) {
socket->reset();
return false;
}
return true;
}
bool sctp_init_client(socket_handler_t* socket,
net_utils::socket_type socktype,
const char* bind_addr_str,
srslte::log* log_)
{
return sctp_init_socket(socket, socktype, bind_addr_str, 0, log_);
}
bool sctp_init_server(socket_handler_t* socket,
net_utils::socket_type socktype,
const char* bind_addr_str,
int port,
srslte::log* log_)
{
if (not sctp_init_socket(socket, socktype, bind_addr_str, port, log_)) {
return false;
}
// Listen for connections
if (listen(sockfd, SOMAXCONN) != 0) {
perror("listen");
return SRSLTE_ERROR;
if (listen(socket->fd(), SOMAXCONN) != 0) {
log_->error("Failed to listen to incoming SCTP connections\n");
return false;
}
return SRSLTE_SUCCESS;
}
int sctp_socket_t::connect_addr(const char* bind_addr_str, const char* dest_addr_str, int dest_port)
{
if (sockfd < 0 and bind_addr(bind_addr_str, 0) != 0) {
reset();
return SRSLTE_ERROR;
}
if (connect_to(&dest_addr, dest_addr_str, dest_port) != 0) {
return SRSLTE_ERROR;
}
return SRSLTE_SUCCESS;
}
int sctp_socket_t::read(void* buf, size_t nbytes, net_addr_t* addr) const
{
if (addr != nullptr) {
sockaddr_in* from = &addr->get_sockaddr_in();
socklen_t fromlen = sizeof(*from);
return read(buf, nbytes, from, &fromlen);
}
return read(buf, nbytes);
}
int sctp_socket_t::read(void* buf,
size_t nbytes,
struct sockaddr_in* from,
socklen_t* fromlen,
struct sctp_sndrcvinfo* sinfo,
int msg_flags) const
{
if (from != nullptr) {
*fromlen = sizeof(*from);
return sctp_recvmsg(sockfd, buf, nbytes, (struct sockaddr*)from, fromlen, sinfo, &msg_flags);
}
return sctp_recvmsg(sockfd, buf, nbytes, nullptr, nullptr, sinfo, &msg_flags);
}
int sctp_socket_t::send(void* buf, size_t nbytes, uint32_t ppid, uint32_t stream_id) const
{
return sctp_sendmsg(
sockfd, buf, nbytes, (struct sockaddr*)&dest_addr, sizeof(dest_addr), htonl(ppid), 0, stream_id, 0, 0);
}
// Private Methods
int sctp_socket_t::create_socket()
{
sockfd = socket(AF_INET, SOCK_SEQPACKET, IPPROTO_SCTP);
if (sockfd == -1) {
perror("Could not create SCTP socket\n");
return -1;
}
return 0;
return true;
}
/***************************************************************
* TCP Socket
**************************************************************/
void tcp_socket_t::reset()
bool tcp_make_server(socket_handler_t* socket,
const char* bind_addr_str,
int port,
int nof_connections,
srslte::log* log_)
{
reset_();
dest_addr = {};
if (connfd >= 0) {
connfd = -1;
if (not socket->open_socket(addr_family::ipv4, socket_type::stream, protocol_type::TCP, log_)) {
return false;
}
}
int tcp_socket_t::listen_addr(const char* bind_addr_str, int port)
{
if (sockfd < 0 and bind_addr(bind_addr_str, port) != 0) {
reset();
return -1;
if (not socket->bind_addr(bind_addr_str, port, log_)) {
socket->reset();
return false;
}
// Listen for connections
if (listen(sockfd, 1) != 0) {
perror("listen");
return -1;
if (listen(socket->fd(), nof_connections) != 0) {
if (log_ != nullptr) {
log_->error("Failed to listen to incoming TCP connections\n");
}
return false;
}
return 0;
return true;
}
int tcp_socket_t::accept_connection()
int tcp_accept(socket_handler_t* socket, sockaddr_in* destaddr, srslte::log* log_)
{
socklen_t clilen = sizeof(dest_addr);
connfd = accept(sockfd, (struct sockaddr*)&dest_addr, &clilen);
socklen_t clilen = sizeof(destaddr);
int connfd = accept(socket->fd(), (struct sockaddr*)&destaddr, &clilen);
if (connfd < 0) {
if (log_ != nullptr) {
log_->error("Failed to accept connection\n");
}
perror("accept");
return -1;
}
return 0;
return connfd;
}
int tcp_socket_t::connect_addr(const char* bind_addr_str, const char* dest_addr_str, int dest_port)
int tcp_read(int remotefd, void* buf, size_t nbytes, srslte::log* log_)
{
if (sockfd < 0 and bind_addr(bind_addr_str, 0) != 0) {
return -1;
}
return connect_to(&dest_addr, dest_addr_str, dest_port);
}
int tcp_socket_t::create_socket()
{
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == -1) {
perror("Could not create TCP socket\n");
return SRSLTE_ERROR;
}
return SRSLTE_SUCCESS;
}
int tcp_socket_t::read(void* buf, size_t nbytes) const
{
int n = ::read(connfd, buf, nbytes);
int n = ::read(remotefd, buf, nbytes);
if (n == 0) {
if (log_ != nullptr) {
log_->info("TCP connection closed\n");
close(remotefd);
}
return 0;
}
if (n == -1) {
perror("read");
if (log_ != nullptr) {
log_->error("Failed to read from TCP socket.");
} else {
perror("TCP read");
}
}
return n;
}
int tcp_socket_t::send(const void* buf, size_t nbytes) const
int tcp_send(int remotefd, const void* buf, size_t nbytes, srslte::log* log_)
{
// Loop until all bytes are sent
char* ptr = (char*)buf;
while (nbytes > 0) {
ssize_t i = ::send(connfd, ptr, nbytes, 0);
char* ptr = (char*)buf;
ssize_t nbytes_remaining = nbytes;
while (nbytes_remaining > 0) {
ssize_t i = ::send(remotefd, ptr, nbytes_remaining, 0);
if (i < 1) {
perror("Error calling send()\n");
return SRSLTE_ERROR;
if (log_ != nullptr) {
log_->error("Failed to send data to TCP socket\n");
} else {
perror("Error calling send()\n");
}
return i;
}
ptr += i;
nbytes -= i;
nbytes_remaining -= i;
}
return SRSLTE_SUCCESS;
return nbytes - nbytes_remaining;
}
} // namespace net_utils
/***************************************************************
* Rx Multisocket Task Types
**************************************************************/
/**
* Description: Specialization of recv_task for the case the received data is
* in the form of unique_byte_buffer, and a recv(...) call is used
*/
class recv_pdu_task final : public rx_multisocket_handler::recv_task
{
public:
using callback_t = std::function<void(srslte::unique_byte_buffer_t pdu)>;
explicit recv_pdu_task(srslte::byte_buffer_pool* pool_, callback_t func_) : pool(pool_), func(std::move(func_)) {}
bool operator()(int fd) override
{
srslte::unique_byte_buffer_t pdu = srslte::allocate_unique_buffer(*pool, "Rxsocket", true);
// inside rx_sockets thread. Read socket
ssize_t n_recv = recv(fd, pdu->msg, pdu->get_tailroom(), 0);
if (n_recv > 0) {
pdu->N_bytes = static_cast<uint32_t>(n_recv);
}
func(std::move(pdu));
return n_recv != 0;
}
private:
srslte::byte_buffer_pool* pool = nullptr;
callback_t func;
};
class sctp_recvmsg_pdu_task final : public rx_multisocket_handler::recv_task
{
public:
using callback_t = std::function<
void(srslte::unique_byte_buffer_t pdu, const sockaddr_in& from, const sctp_sndrcvinfo& sri, int flags)>;
explicit sctp_recvmsg_pdu_task(srslte::byte_buffer_pool* pool_, srslte::log* log_, callback_t func_) :
pool(pool_),
log_h(log_),
func(std::move(func_))
{
}
bool operator()(int fd) override
{
// inside rx_sockets thread. Read socket
srslte::unique_byte_buffer_t pdu = srslte::allocate_unique_buffer(*pool, "Rxsocket", true);
sockaddr_in from = {};
socklen_t fromlen = sizeof(from);
sctp_sndrcvinfo sri = {};
int flags = 0;
ssize_t n_recv = sctp_recvmsg(fd, pdu->msg, pdu->get_tailroom(), (struct sockaddr*)&from, &fromlen, &sri, &flags);
if (n_recv == -1 and errno != EAGAIN) {
log_h->error("Error reading from SCTP socket: %s\n", strerror(errno));
return true;
}
if (n_recv == -1 and errno == EAGAIN) {
log_h->debug("Socket timeout reached\n");
return true;
}
bool ret = true;
pdu->N_bytes = static_cast<uint32_t>(n_recv);
if (flags & MSG_NOTIFICATION) {
// Received notification
union sctp_notification* notification = (union sctp_notification*)pdu->msg;
if (notification->sn_header.sn_type == SCTP_SHUTDOWN_EVENT) {
// Socket Shutdown
ret = false;
}
}
func(std::move(pdu), from, sri, flags);
return ret;
}
private:
srslte::byte_buffer_pool* pool = nullptr;
srslte::log* log_h = nullptr;
callback_t func;
};
/***************************************************************
* Rx Multisocket Handler
**************************************************************/
rx_multisocket_handler::rx_multisocket_handler(std::string name_, srslte::log* log_) :
rx_multisocket_handler::rx_multisocket_handler(std::string name_, srslte::log* log_, int thread_prio) :
thread(name_),
name(std::move(name_)),
log_h(log_)
{
pool = srslte::byte_buffer_pool::get_instance();
// register control pipe fd
if (pipe(pipefd) == -1) {
rxSockInfo("Failed to open control pipe\n");
return;
}
start(THREAD_PRIO);
start(thread_prio);
}
rx_multisocket_handler::~rx_multisocket_handler()
{
if (running) {
std::lock_guard<std::mutex> lock(socket_mutex);
ctrl_cmd_t msg{};
msg.cmd = ctrl_cmd_t::cmd_id_t::EXIT;
rxSockDebug("Closing socket handler\n");
if (write(pipefd[1], &msg, sizeof(msg)) != sizeof(msg)) {
rxSockError("while writing to control pipe\n");
}
}
// close thread
wait_thread_finish();
close(pipefd[0]);
close(pipefd[1]);
rxSockDebug("closed.\n");
stop();
}
bool rx_multisocket_handler::register_socket_(std::pair<const int, std::function<void()> >&& elem)
void rx_multisocket_handler::stop()
{
if (running) {
// close thread
{
std::lock_guard<std::mutex> lock(socket_mutex);
ctrl_cmd_t msg{};
msg.cmd = ctrl_cmd_t::cmd_id_t::EXIT;
if (write(pipefd[1], &msg, sizeof(msg)) != sizeof(msg)) {
rxSockError("while writing to control pipe\n");
}
}
rxSockDebug("Closing rx socket handler thread\n");
wait_thread_finish();
}
if (pipefd[0] >= 0) {
close(pipefd[0]);
close(pipefd[1]);
pipefd[0] = -1;
pipefd[1] = -1;
rxSockDebug("closed.\n");
}
}
/**
* Convenience method for read PDUs from socket
*/
bool rx_multisocket_handler::add_socket_pdu_handler(int fd, recv_callback_t pdu_task)
{
std::unique_ptr<srslte::rx_multisocket_handler::recv_task> task;
task.reset(new srslte::recv_pdu_task(pool, std::move(pdu_task)));
return add_socket_handler(fd, std::move(task));
}
/**
* Convenience method for reading PDUs from SCTP socket
*/
bool rx_multisocket_handler::add_socket_sctp_handler(int fd, sctp_recv_callback_t pdu_task)
{
srslte::rx_multisocket_handler::task_callback_t task;
task.reset(new srslte::sctp_recvmsg_pdu_task(pool, log_h, std::move(pdu_task)));
return add_socket_handler(fd, std::move(task));
}
bool rx_multisocket_handler::add_socket_handler(int fd, task_callback_t handler)
{
int fd = elem.first;
std::lock_guard<std::mutex> lock(socket_mutex);
if (fd < 0) {
rxSockError("Provided SCTP socket must be already open\n");
@ -357,7 +504,7 @@ bool rx_multisocket_handler::register_socket_(std::pair<const int, std::function
return false;
}
active_sockets.insert(std::move(elem));
active_sockets.insert(std::pair<const int, task_callback_t>(fd, std::move(handler)));
// this unlocks the reading thread to add new connections
ctrl_cmd_t msg;
@ -365,12 +512,46 @@ bool rx_multisocket_handler::register_socket_(std::pair<const int, std::function
msg.new_fd = fd;
if (write(pipefd[1], &msg, sizeof(msg)) != sizeof(msg)) {
rxSockError("while writing to control pipe\n");
return false;
}
rxSockDebug("socket fd=%d has been registered.\n", fd);
return true;
}
bool rx_multisocket_handler::remove_socket(int fd)
{
std::lock_guard<std::mutex> lock(socket_mutex);
auto it = active_sockets.find(fd);
if (it == active_sockets.end()) {
rxSockError("The socket fd=%d to be removed does not exist\n", fd);
return false;
}
ctrl_cmd_t msg;
msg.cmd = ctrl_cmd_t::cmd_id_t::RM_FD;
msg.new_fd = fd;
if (write(pipefd[1], &msg, sizeof(msg)) != sizeof(msg)) {
rxSockError("while writing to control pipe\n");
return false;
}
return true;
}
bool rx_multisocket_handler::remove_socket_unprotected(int fd, fd_set* total_fd_set, int* max_fd)
{
if (fd < 0) {
rxSockError("fd to be removed is not valid\n");
return false;
}
active_sockets.erase(fd);
FD_CLR(fd, total_fd_set);
// assumes ordering
*max_fd = (active_sockets.empty()) ? pipefd[0] : std::max(pipefd[0], active_sockets.rbegin()->first);
rxSockDebug("Socket fd=%d has been successfully removed\n", fd);
return true;
}
void rx_multisocket_handler::run_thread()
{
running = true;
@ -387,7 +568,7 @@ void rx_multisocket_handler::run_thread()
// handle select return
if (n == -1) {
rxSockError("Error from select()");
rxSockError("Error from select(%d,...). Number of rx sockets: %d", max_fd + 1, (int)active_sockets.size() + 1);
continue;
}
if (n == 0) {
@ -395,6 +576,7 @@ void rx_multisocket_handler::run_thread()
continue;
}
// Shared state area
std::lock_guard<std::mutex> lock(socket_mutex);
// call read callback for all SCTP/TCP/UDP connections
@ -402,7 +584,11 @@ void rx_multisocket_handler::run_thread()
if (not FD_ISSET(handler_pair.first, &read_fd_set)) {
continue;
}
handler_pair.second();
bool socket_valid = (*handler_pair.second)(handler_pair.first);
if (not socket_valid) {
rxSockWarn("The socket fd=%d has been closed by peer\n", handler_pair.first);
remove_socket_unprotected(handler_pair.first, &total_fd_set, &max_fd);
}
}
// handle ctrl messages
@ -425,6 +611,10 @@ void rx_multisocket_handler::run_thread()
rxSockError("added fd is not valid\n");
}
break;
case ctrl_cmd_t::cmd_id_t::RM_FD:
remove_socket_unprotected(msg.new_fd, &total_fd_set, &max_fd);
rxSockDebug("Socket fd=%d has been successfully removed\n", msg.new_fd);
break;
default:
rxSockError("ctrl message command %d is not valid\n", (int)msg.cmd);
}

2
lib/test/common/CMakeLists.txt
View File

@ -78,6 +78,8 @@ add_test(queue_test queue_test)
add_executable(timer_test timer_test.cc)
target_link_libraries(timer_test srslte_common)
add_test(timer_test timer_test)
add_executable(network_utils_test network_utils_test.cc)
target_link_libraries(network_utils_test srslte_common ${CMAKE_THREAD_LIBS_INIT} ${SCTP_LIBRARIES})
add_test(network_utils_test network_utils_test)

62
lib/test/common/network_utils_test.cc
View File

@ -37,44 +37,56 @@ int test_socket_handler()
log.set_level(srslte::LOG_LEVEL_DEBUG);
log.set_hex_limit(128);
int counter = 0;
srslte::byte_buffer_pool* pool = srslte::byte_buffer_pool::get_instance();
int counter = 0;
srslte::sctp_socket_t server_sock, client_sock, client_sock2;
srslte::socket_handler_t server_socket, client_socket, client_socket2;
srslte::rx_multisocket_handler sockhandler("RXSOCKETS", &log);
int server_port = 36412;
const char* server_addr = "127.0.100.1";
using namespace srslte::net_utils;
TESTASSERT(server_sock.listen_addr("127.0.100.1", 36412) == 0);
log.info("Listening from fd=%d\n", server_sock.fd());
TESTASSERT(sctp_init_server(&server_socket, socket_type::seqpacket, server_addr, server_port, &log));
log.info("Listening from fd=%d\n", server_socket.fd());
TESTASSERT(client_sock.connect_addr("127.0.0.1", "127.0.100.1", 36412) == 0);
TESTASSERT(client_sock2.connect_addr("127.0.0.2", "127.0.100.1", 36412) == 0);
TESTASSERT(sctp_init_client(&client_socket, socket_type::seqpacket, "127.0.0.1", &log));
TESTASSERT(sctp_init_client(&client_socket2, socket_type::seqpacket, "127.0.0.2", &log));
TESTASSERT(client_socket.connect_to(server_addr, server_port));
TESTASSERT(client_socket2.connect_to(server_addr, server_port));
// register server Rx handler
sockhandler.register_socket(server_sock, [pool, &log, &counter](const srslte::sctp_socket_t& sock) {
srslte::unique_byte_buffer_t pdu = srslte::allocate_unique_buffer(*pool, true);
srslte::net_addr_t addr;
int rd_sz = sock.read(pdu->msg, pdu->get_tailroom(), &addr);
if (rd_sz > 0) {
pdu->N_bytes = rd_sz;
log.info_hex(pdu->msg, pdu->N_bytes, "Received msg from %s:", addr.ip().c_str());
counter++;
}
});
auto pdu_handler =
[&log,
&counter](srslte::unique_byte_buffer_t pdu, const sockaddr_in& from, const sctp_sndrcvinfo& sri, int flags) {
if (pdu->N_bytes > 0) {
log.info_hex(pdu->msg, pdu->N_bytes, "Received msg from %s:", get_ip(from).c_str());
counter++;
}
};
sockhandler.add_socket_sctp_handler(server_socket.fd(), pdu_handler);
int PPID = 18;
const int NONUE_STREAM_ID = 0;
uint8_t buf[128] = {};
int32_t nof_counts = 5;
uint8_t buf[128] = {};
int32_t nof_counts = 5;
sockaddr_in server_addrin = server_socket.get_addr_in();
socklen_t socklen = sizeof(server_addrin);
const int NONUE_STREAM_ID = 0;
for (int32_t i = 0; i < nof_counts; ++i) {
buf[i] = i;
// Round-robin between clients
srslte::sctp_socket_t* chosen = &client_sock;
srslte::socket_handler_t* chosen = &client_socket;
if (i % 2 == 1) {
chosen = &client_sock2;
chosen = &client_socket2;
}
// send packet
ssize_t n_sent = chosen->send(buf, i + 1, PPID, NONUE_STREAM_ID);
ssize_t n_sent = sctp_sendmsg(chosen->fd(),
buf,
i + 1,
(struct sockaddr*)&server_addrin,
socklen,
(uint32_t)ppid_values::S1AP,
0,
NONUE_STREAM_ID,
0,
0);
TESTASSERT(n_sent >= 0);
usleep(1000);
log.info("Message %d sent.\n", i);

32
srsenb/hdr/stack/enb_stack_lte.h
View File

@ -43,7 +43,10 @@
namespace srsenb {
class enb_stack_lte final : public enb_stack_base, public stack_interface_phy_lte, public thread
class enb_stack_lte final : public enb_stack_base,
public stack_interface_phy_lte,
public stack_interface_s1ap_lte,
public thread
{
public:
enb_stack_lte(srslte::logger* logger_);
@ -52,7 +55,7 @@ public:
// eNB stack base interface
int init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_, phy_interface_stack_lte* phy_);
int init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_);
void stop() final;
void stop() final;
std::string get_type() final;
bool get_metrics(stack_metrics_t* metrics) final;
@ -79,7 +82,7 @@ public:
{
return mac.get_mch_sched(tti, is_mcch, dl_sched_res);
}
int get_ul_sched(uint32_t tti, ul_sched_t* ul_sched_res) final { return mac.get_ul_sched(tti, ul_sched_res); }
int get_ul_sched(uint32_t tti, ul_sched_t* ul_sched_res) final { return mac.get_ul_sched(tti, ul_sched_res); }
void set_sched_dl_tti_mask(uint8_t* tti_mask, uint32_t nof_sfs) final
{
mac.set_sched_dl_tti_mask(tti_mask, nof_sfs);
@ -89,18 +92,28 @@ public:
void rl_ok(uint16_t rnti) final { mac.rl_ok(rnti); }
void tti_clock() override;
/* STACK-S1AP interface*/
void add_mme_socket(int fd) override;
void remove_mme_socket(int fd) override;
private:
static const int STACK_MAIN_THREAD_PRIO = -1; // Use default high-priority below UHD
// thread loop
void run_thread() override;
void stop_impl();
void tti_clock_impl();
void handle_mme_rx_packet(srslte::unique_byte_buffer_t pdu,
const sockaddr_in& from,
const sctp_sndrcvinfo& sri,
int flags);
// args
stack_args_t args = {};
rrc_cfg_t rrc_cfg = {};
srslte::timer_handler timers;
// components that layers depend on (need to be destroyed after layers)
srslte::timer_handler timers;
std::unique_ptr<srslte::rx_multisocket_handler> rx_sockets;
srsenb::mac mac;
srslte::mac_pcap mac_pcap;
@ -110,7 +123,8 @@ private:
srsenb::gtpu gtpu;
srsenb::s1ap s1ap;
srslte::logger* logger = nullptr;
srslte::logger* logger = nullptr;
srslte::byte_buffer_pool* pool = nullptr;
// Radio and PHY log are in enb.cc
srslte::log_filter mac_log;
@ -119,6 +133,7 @@ private:
srslte::log_filter rrc_log;
srslte::log_filter s1ap_log;
srslte::log_filter gtpu_log;
srslte::log_filter stack_log;
// RAT-specific interfaces
phy_interface_stack_lte* phy = nullptr;
@ -131,10 +146,15 @@ private:
srslte::unique_byte_buffer_t pdu;
task_t() = default;
explicit task_t(std::function<void(task_t*)> f_) : func(std::move(f_)) {}
task_t(std::function<void(task_t*)> f_, srslte::unique_byte_buffer_t pdu_) :
func(std::move(f_)),
pdu(std::move(pdu_))
{
}
void operator()() { func(this); }
};
srslte::multiqueue_handler<task_t> pending_tasks;
int enb_queue_id = -1, sync_queue_id = -1;
int enb_queue_id = -1, sync_queue_id = -1, mme_queue_id = -1;
};
} // namespace srsenb

111
srsenb/hdr/stack/upper/s1ap.h
View File

@ -24,21 +24,22 @@
#include <map>
#include "common_enb.h"
#include "srslte/common/buffer_pool.h"
#include "srslte/common/log.h"
#include "srslte/common/common.h"
#include "srslte/common/log.h"
#include "srslte/common/threads.h"
#include "srslte/interfaces/enb_interfaces.h"
#include "common_enb.h"
#include "s1ap_metrics.h"
#include "srslte/asn1/liblte_s1ap.h"
#include "srslte/common/network_utils.h"
#include "srslte/common/stack_procedure.h"
namespace srsenb {
typedef struct {
uint32_t rnti;
uint32_t rnti;
uint32_t eNB_UE_S1AP_ID;
uint32_t MME_UE_S1AP_ID;
bool release_requested;
@ -46,16 +47,18 @@ typedef struct {
struct timeval init_timestamp;
} ue_ctxt_t;
class s1ap : public s1ap_interface_rrc, public thread
class s1ap : public s1ap_interface_rrc
{
public:
s1ap();
bool init(s1ap_args_t args_, rrc_interface_s1ap* rrc_, srslte::log* s1ap_log_, srslte::timer_handler* timers_);
bool init(s1ap_args_t args_,
rrc_interface_s1ap* rrc_,
srslte::log* s1ap_log_,
srslte::timer_handler* timers_,
srsenb::stack_interface_s1ap_lte* stack_);
void stop();
void get_metrics(s1ap_metrics_t& m);
void run_thread();
// RRC interface
void initial_ue(uint16_t rnti, LIBLTE_S1AP_RRC_ESTABLISHMENT_CAUSE_ENUM cause, srslte::unique_byte_buffer_t pdu);
void initial_ue(uint16_t rnti,
@ -64,33 +67,40 @@ public:
uint32_t m_tmsi,
uint8_t mmec);
void write_pdu(uint16_t rnti, srslte::unique_byte_buffer_t pdu);
bool user_exists(uint16_t rnti);
bool user_exists(uint16_t rnti);
bool user_release(uint16_t rnti, LIBLTE_S1AP_CAUSERADIONETWORK_ENUM cause_radio);
void ue_ctxt_setup_complete(uint16_t rnti, LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPRESPONSE_STRUCT *res);
void ue_erab_setup_complete(uint16_t rnti, LIBLTE_S1AP_MESSAGE_E_RABSETUPRESPONSE_STRUCT *res);
void ue_ctxt_setup_complete(uint16_t rnti, LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPRESPONSE_STRUCT* res);
void ue_erab_setup_complete(uint16_t rnti, LIBLTE_S1AP_MESSAGE_E_RABSETUPRESPONSE_STRUCT* res);
bool is_mme_connected();
//void ue_capabilities(uint16_t rnti, LIBLTE_RRC_UE_EUTRA_CAPABILITY_STRUCT *caps);
// void ue_capabilities(uint16_t rnti, LIBLTE_RRC_UE_EUTRA_CAPABILITY_STRUCT *caps);
// Stack interface
// bool handle_mme_rx_msg(srslte::unique_byte_buffer_t pdu);
bool
handle_mme_rx_msg(srslte::unique_byte_buffer_t pdu, const sockaddr_in& from, const sctp_sndrcvinfo& sri, int flags);
private:
static const int S1AP_THREAD_PRIO = 65;
static const int MME_PORT = 36412;
static const int ADDR_FAMILY = AF_INET;
static const int SOCK_TYPE = SOCK_STREAM;
static const int PROTO = IPPROTO_SCTP;
static const int PPID = 18;
static const int NONUE_STREAM_ID = 0;
static const int MME_PORT = 36412;
static const int ADDR_FAMILY = AF_INET;
static const int SOCK_TYPE = SOCK_STREAM;
static const int PROTO = IPPROTO_SCTP;
static const int PPID = 18;
static const int NONUE_STREAM_ID = 0;
rrc_interface_s1ap *rrc;
s1ap_args_t args;
srslte::log *s1ap_log;
srslte::byte_buffer_pool *pool;
// args
rrc_interface_s1ap* rrc = nullptr;
s1ap_args_t args;
srslte::log* s1ap_log = nullptr;
srslte::byte_buffer_pool* pool = nullptr;
srsenb::stack_interface_s1ap_lte* stack = nullptr;
bool mme_connected;
bool running;
int socket_fd; // SCTP socket file descriptor
struct sockaddr_in mme_addr; // MME address
uint32_t next_eNB_UE_S1AP_ID; // Next ENB-side UE identifier
uint16_t next_ue_stream_id; // Next UE SCTP stream identifier
srslte::socket_handler_t s1ap_socket;
struct sockaddr_in mme_addr = {}; // MME address
bool mme_connected = false;
bool running = false;
uint32_t next_eNB_UE_S1AP_ID = 1; // Next ENB-side UE identifier
uint16_t next_ue_stream_id = 1; // Next UE SCTP stream identifier
srslte::timer_handler::unique_timer mme_connect_timer, s1setup_timeout;
// Protocol IEs sent with every UL S1AP message
LIBLTE_S1AP_TAI_STRUCT tai;
@ -106,16 +116,16 @@ private:
bool handle_s1ap_rx_pdu(srslte::byte_buffer_t* pdu);
bool handle_initiatingmessage(LIBLTE_S1AP_INITIATINGMESSAGE_STRUCT* msg);
bool handle_successfuloutcome(LIBLTE_S1AP_SUCCESSFULOUTCOME_STRUCT* msg);
bool handle_unsuccessfuloutcome(LIBLTE_S1AP_UNSUCCESSFULOUTCOME_STRUCT *msg);
bool handle_paging(LIBLTE_S1AP_MESSAGE_PAGING_STRUCT *msg);
bool handle_unsuccessfuloutcome(LIBLTE_S1AP_UNSUCCESSFULOUTCOME_STRUCT* msg);
bool handle_paging(LIBLTE_S1AP_MESSAGE_PAGING_STRUCT* msg);
bool handle_s1setupresponse(LIBLTE_S1AP_MESSAGE_S1SETUPRESPONSE_STRUCT *msg);
bool handle_dlnastransport(LIBLTE_S1AP_MESSAGE_DOWNLINKNASTRANSPORT_STRUCT *msg);
bool handle_initialctxtsetuprequest(LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPREQUEST_STRUCT *msg);
bool handle_uectxtreleasecommand(LIBLTE_S1AP_MESSAGE_UECONTEXTRELEASECOMMAND_STRUCT *msg);
bool handle_s1setupfailure(LIBLTE_S1AP_MESSAGE_S1SETUPFAILURE_STRUCT *msg);
bool handle_erabsetuprequest(LIBLTE_S1AP_MESSAGE_E_RABSETUPREQUEST_STRUCT *msg);
bool handle_uecontextmodifyrequest(LIBLTE_S1AP_MESSAGE_UECONTEXTMODIFICATIONREQUEST_STRUCT *msg);
bool handle_s1setupresponse(LIBLTE_S1AP_MESSAGE_S1SETUPRESPONSE_STRUCT* msg);
bool handle_dlnastransport(LIBLTE_S1AP_MESSAGE_DOWNLINKNASTRANSPORT_STRUCT* msg);
bool handle_initialctxtsetuprequest(LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPREQUEST_STRUCT* msg);
bool handle_uectxtreleasecommand(LIBLTE_S1AP_MESSAGE_UECONTEXTRELEASECOMMAND_STRUCT* msg);
bool handle_s1setupfailure(LIBLTE_S1AP_MESSAGE_S1SETUPFAILURE_STRUCT* msg);
bool handle_erabsetuprequest(LIBLTE_S1AP_MESSAGE_E_RABSETUPREQUEST_STRUCT* msg);
bool handle_uecontextmodifyrequest(LIBLTE_S1AP_MESSAGE_UECONTEXTMODIFICATIONREQUEST_STRUCT* msg);
bool send_initialuemessage(uint16_t rnti,
LIBLTE_S1AP_RRC_ESTABLISHMENT_CAUSE_ENUM cause,
@ -124,9 +134,9 @@ private:
uint32_t m_tmsi = 0,
uint8_t mmec = 0);
bool send_ulnastransport(uint16_t rnti, srslte::unique_byte_buffer_t pdu);
bool send_uectxtreleaserequest(uint16_t rnti, LIBLTE_S1AP_CAUSE_STRUCT *cause);
bool send_uectxtreleaserequest(uint16_t rnti, LIBLTE_S1AP_CAUSE_STRUCT* cause);
bool send_uectxtreleasecomplete(uint16_t rnti, uint32_t mme_ue_id, uint32_t enb_ue_id);
bool send_initial_ctxt_setup_response(uint16_t rnti, LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPRESPONSE_STRUCT *res_);
bool send_initial_ctxt_setup_response(uint16_t rnti, LIBLTE_S1AP_MESSAGE_INITIALCONTEXTSETUPRESPONSE_STRUCT* res_);
bool send_initial_ctxt_setup_failure(uint16_t rnti);
bool send_erab_setup_response(uint16_t rnti, LIBLTE_S1AP_MESSAGE_E_RABSETUPRESPONSE_STRUCT* res_);
// bool send_ue_capabilities(uint16_t rnti, LIBLTE_RRC_UE_EUTRA_CAPABILITY_STRUCT *caps)
@ -191,6 +201,29 @@ private:
// timers
srslte::timer_handler* timers = nullptr;
// procedures
class s1_setup_proc_t
{
public:
struct s1setupresult {
bool success = false;
enum class cause_t { timeout, failure } cause;
};
explicit s1_setup_proc_t(s1ap* s1ap_) : s1ap_ptr(s1ap_) {}
srslte::proc_outcome_t init();
srslte::proc_outcome_t step() { return srslte::proc_outcome_t::yield; }
srslte::proc_outcome_t react(const s1setupresult& event);
void then(const srslte::proc_state_t& result) const;
const char* name() const { return "MME Connection"; }
private:
srslte::proc_outcome_t start_mme_connection();
s1ap* s1ap_ptr = nullptr;
};
srslte::proc_t<s1_setup_proc_t> s1setup_proc;
};
} // namespace srsenb

56
srsenb/src/stack/enb_stack_lte.cc
View File

@ -21,6 +21,7 @@
#include "srsenb/hdr/stack/enb_stack_lte.h"
#include "srsenb/hdr/enb.h"
#include "srslte/common/network_utils.h"
#include "srslte/srslte.h"
#include <srslte/interfaces/enb_metrics_interface.h>
@ -30,6 +31,11 @@ namespace srsenb {
enb_stack_lte::enb_stack_lte(srslte::logger* logger_) : logger(logger_), pdcp(&pdcp_log), timers(128), thread("STACK")
{
enb_queue_id = pending_tasks.add_queue();
sync_queue_id = pending_tasks.add_queue();
mme_queue_id = pending_tasks.add_queue();
pool = byte_buffer_pool::get_instance();
}
enb_stack_lte::~enb_stack_lte()
@ -64,6 +70,7 @@ int enb_stack_lte::init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_)
rrc_log.init("RRC ", logger);
gtpu_log.init("GTPU", logger);
s1ap_log.init("S1AP", logger);
stack_log.init("STACK", logger);
// Init logs
mac_log.set_level(args.log.mac_level);
@ -72,6 +79,7 @@ int enb_stack_lte::init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_)
rrc_log.set_level(args.log.rrc_level);
gtpu_log.set_level(args.log.gtpu_level);
s1ap_log.set_level(args.log.s1ap_level);
stack_log.set_level("INFO");
mac_log.set_hex_limit(args.log.mac_hex_limit);
rlc_log.set_hex_limit(args.log.rlc_hex_limit);
@ -79,6 +87,7 @@ int enb_stack_lte::init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_)
rrc_log.set_hex_limit(args.log.rrc_hex_limit);
gtpu_log.set_hex_limit(args.log.gtpu_hex_limit);
s1ap_log.set_hex_limit(args.log.s1ap_hex_limit);
stack_log.set_hex_limit(128);
// Set up pcap and trace
if (args.pcap.enable) {
@ -116,12 +125,15 @@ int enb_stack_lte::init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_)
}
}
// Init Rx socket handler
rx_sockets.reset(new srslte::rx_multisocket_handler("ENBSOCKETS", &stack_log));
// Init all layers
mac.init(args.mac, &cell_cfg, phy, &rlc, &rrc, &mac_log);
rlc.init(&pdcp, &rrc, &mac, &timers, &rlc_log);
pdcp.init(&rlc, &rrc, &gtpu);
rrc.init(&rrc_cfg, phy, &mac, &rlc, &pdcp, &s1ap, &gtpu, &timers, &rrc_log);
s1ap.init(args.s1ap, &rrc, &s1ap_log, &timers);
s1ap.init(args.s1ap, &rrc, &s1ap_log, &timers, this);
gtpu.init(args.s1ap.gtp_bind_addr,
args.s1ap.mme_addr,
args.embms.m1u_multiaddr,
@ -130,9 +142,6 @@ int enb_stack_lte::init(const stack_args_t& args_, const rrc_cfg_t& rrc_cfg_)
&gtpu_log,
args.embms.enable);
enb_queue_id = pending_tasks.add_queue();
sync_queue_id = pending_tasks.add_queue();
started = true;
start(STACK_MAIN_THREAD_PRIO);
@ -161,9 +170,6 @@ void enb_stack_lte::stop()
void enb_stack_lte::stop_impl()
{
// stop listening to events
pending_tasks.erase_queue(sync_queue_id);
pending_tasks.erase_queue(enb_queue_id);
s1ap.stop();
gtpu.stop();
@ -178,6 +184,14 @@ void enb_stack_lte::stop_impl()
if (args.pcap.enable) {
mac_pcap.close();
}
rx_sockets->stop();
// erasing the queues is the last thing, bc we need them to call stop_impl()
pending_tasks.erase_queue(sync_queue_id);
pending_tasks.erase_queue(enb_queue_id);
pending_tasks.erase_queue(mme_queue_id);
started = false;
}
@ -199,4 +213,32 @@ void enb_stack_lte::run_thread()
}
}
void enb_stack_lte::handle_mme_rx_packet(srslte::unique_byte_buffer_t pdu,
const sockaddr_in& from,
const sctp_sndrcvinfo& sri,
int flags)
{
// Defer the handling of MME packet to eNB stack main thread
auto task_handler = [this, from, sri, flags](task_t* t) {
s1ap.handle_mme_rx_msg(std::move(t->pdu), from, sri, flags);
};
// Defer the handling of MME packet to main stack thread
pending_tasks.push(mme_queue_id, task_t{task_handler, std::move(pdu)});
}
void enb_stack_lte::add_mme_socket(int fd)
{
// Pass MME Rx packet handler functor to socket handler to run in socket thread
auto mme_rx_handler =
[this](srslte::unique_byte_buffer_t pdu, const sockaddr_in& from, const sctp_sndrcvinfo& sri, int flags) {
handle_mme_rx_packet(std::move(pdu), from, sri, flags);
};
rx_sockets->add_socket_sctp_handler(fd, mme_rx_handler);
}
void enb_stack_lte::remove_mme_socket(int fd)
{
rx_sockets->remove_socket(fd);
}
} // namespace srsenb

5
srsenb/src/stack/rrc/rrc.cc
View File

@ -40,7 +40,6 @@ namespace srsenb {
rrc::rrc() : cnotifier(nullptr), nof_si_messages(0)
{
users.clear();
pending_paging.clear();
}
@ -206,7 +205,7 @@ void rrc::add_user(uint16_t rnti)
pthread_mutex_lock(&user_mutex);
auto user_it = users.find(rnti);
if (user_it == users.end()) {
users[rnti].reset(new ue{this, rnti});
users.insert(std::make_pair(rnti, std::unique_ptr<ue>(new ue{this, rnti})));
rlc->add_user(rnti);
pdcp->add_user(rnti);
rrc_log->info("Added new user rnti=0x%x\n", rnti);
@ -620,7 +619,7 @@ void rrc::read_pdu_pcch(uint8_t* payload, uint32_t buffer_size)
void rrc::ho_preparation_complete(uint16_t rnti, bool is_success)
{
users[rnti]->handle_ho_preparation_complete(is_success);
users.at(rnti)->handle_ho_preparation_complete(is_success);
}
/*******************************************************************************

849
srsenb/src/stack/upper/s1ap.cc
View File

File diff suppressed because it is too large Load Diff